Building Golf Radio from our Real-Time Insights Engine

[embed]

In 2023, we were tasked with creating an engaging live commentary bot to enhance the watching experience. Leveraging the exclusive strokes-gained statistics we generate for every shot, we were able to identify and highlight exciting moments as they unfolded during live play. Using Apache Druid we could also reference historical data for these events, providing a broader context and gauging their significance across different timescales. Additionally, by integrating rich course features into our models, we were able to anticipate upcoming highlights and pinpoint compelling hole-and-player pairings before they happened.

Stateful Spark

To start we wrote a spark streaming job to collect an entire players tournament into one json state where we could sum any stat that we wanted to use to measure the excitement of what just happened. See this blog on writing a stateful spark streaming job:

[embed]Spark Structured Streaming under 3 seconds latency for the Ryder Cup In this article I’ll take you through the transformations and latency optimisations I learnt whilst building a spark…patrick-green-78196.medium.com

Using this spark streaming job we emit a large json state + stats per player and used a confluent cloud UDF to query Druid. The query based on the information given can then decide how exciting what just happened is and also the exact text comment to send to the US Open website via Kafka. To run confluent UDFs we hosted our own ksqlDB instance with helm and configured it to use our cloud Kafka brokers:

[embed]Deploy Confluent for Kubernetes | Confluent Documentation Learn how to install Confluent for Kubernetes.docs.confluent.io

To make the UDF available we built both the jar and our own docker image, the config we used is below:

FROM confluentinc/cp-ksqldb-server:7.5.1
USER root

RUN mkdir /usr/share/java/ourudfs/
COPY ./target/our-confluent-utility-1.0-jar-with-dependencies.jar /usr/share/java/ourudfs

RUN chown -R appuser.appuser /usr/share/java/ourudfs/
RUN chmod -R 755 /usr/share/java/ourudfs/

USER 1001

import io.confluent.ksql.function.udf.Udf;
import io.confluent.ksql.function.udf.UdfDescription;
import io.confluent.ksql.function.udf.UdfParameter;

@UdfDescription(name = "Commentary", description = "call commentary")
public class Commentary {

  @Udf(description = "calls commentary query endpoint")
  public string call(@UdfParameter(value="tournament") final string tournament, ...) {
    return ....;
  }
}

apiVersion: platform.confluent.io/v1beta1
kind: KsqlDB
metadata:
  name: ksqldb
  namespace: dataops-confluent
spec:
  replicas: 2
  oneReplicaPerNode: true
  image:
    application: [IMAGE-HOSTING-JAR-ABOVE]
    init: confluentinc/confluent-init-container:2.7.1
  license:
    globalLicense: true
  dataVolumeCapacity: 100Gi
  configOverrides:
    server:
      - ksql.service.id=ksqldb
      - ksql.logging.processing.topic.auto.create=true
      - ksql.logging.processing.topic.name=[LOG TOPIC NAME]
      - ksql.logging.processing.topic.partitions=1
      - ksql.logging.processing.topic.replication.factor=3
      - ksql.logging.processing.stream.auto.create=true
      - ksql.logging.processing.stream.name=PROCESSING_LOG
      - ksql.heartbeat.enable=true
      - ksql.internal.topic.replicas=-1
      - ksql.streams.num.stream.threads=4
      - ksql.extension.dir=/usr/share/java/ourudfs/   # The directory in which ksqlDB looks for UDFs
    jvm:
      - "-Xmx16g"
      - "-XX:+UseG1GC"
  dependencies:
    kafka:
      bootstrapEndpoint: [CLOUD BROKER URL]
      authentication:
        type: plain
        jaasConfig:
          secretRef: ####
      tls:
        enabled: true
        ignoreTrustStoreConfig: true
    schemaRegistry:
      url: [CLOUD SCHEMA REGISTRY URL]
      authentication:
        type: basic
        basic:
          secretRef: ####
      tls:
        enabled: true
        ignoreTrustStoreConfig: true

  podTemplate:
    resources:
      requests:
        memory: "8Gi"
      limits:
        memory: "8Gi"

Designing a Commentary Table

To reduce the load on druid we built our own materialised view specifically for commentary to hold the existing top stats for every timeframe and for every hole. By creating one row per hole and per timeframe we could easily reference the significance of a live packet and build an insightful comment.

There are four types columns we stored:

• Basic course information including number of trees, size of fairway and other hole based features
• One hot encoding of whether that hole ranks the highest, lowest or runner up for any particular stat
• Average and standard deviation so we can evaluate the statistical significance of any particular hole and player stat
• The upper and lower bound of a particular hole and player stat including the runner up values so the top 3 and top 10 can be commented on

To calculate this table we ran a MSQ job that reads the current commentary tables top stats that relate to the tournaments that is live and then compares the values to the current tournament and upserts those rows and records that have been broken. Some example insights from the top stats are shown below:

Druid SQL Features

There we’re a few core Druid features that helped develop the query and improved it’s performance:

• Using lookups and window functions we were able to host a cdf lookup to calculate where each stat lay on a standard normal distribution to measure its significance.

• Also using live kafka lookups we we’re able to ingest more complex insights that couldn’t be derived with just SQL. For example we asynchronously run a regression model to infer the difficulty of specific shot on a individual hole by continuously fitting the regression model to our shot by shot strokes-gained statistics. The difficulty of each shot on each hole is then saved to the commentary materialised view during the MSQ job.

• Another feature of druid that we found useful was GROUPING SETS which made it easy to build the different time frames and context for each record in one single query for example ‘today’ , ‘at this course’.

GROUP BY GROUPING SETS (
  (stat_name,stat_value,tournamentId,roundNumber,courseId,holeNo,hole_par),
  (stat_name,stat_value,tournamentId,courseId,holeNo,hole_par),
  (stat_name,stat_value,courseId)
)

• We also had to enable correct null handling. After turning this feature on, the SQL was simplified as we could tell the difference between and 0 stat and a missing stat.

druid.generic.useDefaultValueForNull: "false"

Druid Merge Buffers Performance

Occasionally the large top stat queries would fail. To solve this we updated the broker config to allow a larger memory spill to disk. An example of the config change is below:

druid.query.groupBy.maxOnDiskStorage: "8000000000"
druid.query.groupBy.maxMergingDictionarySize: "2000000000"

We also came across a bug where on some large queries merge buffers weren’t available. This can happen when too many query threads have been setup for the resource available. The suggested amount is one per CPU however with middlemanager ingestors using the same resource not every CPU will be available all of the time.

To solve this issue we updated the historical config to allocate more memory to each merge buffer. The best way to do this is to decrease the number of processing threads druid.processing.numThreads so theirs a larger share of merge buffers per query and increase the historical java heap memory above the general role of thumb thats 0.5 GB per CPU core:

DRUID_MAXDIRECTMEMORYSIZE: "6500m" # java max direct memory
DRUID_XMS: "2000m" # initial heap size
DRUID_XMX: "9000m" # max heap size

Initialising the Commentary Table

To build the initial top stat values we had to run one huge query over 20 years of golf data. For this we used Druid deep storage query feature. For the feature to work added the following configs for durable storage in order to store the results from the query:

  druid.msq.intermediate.storage.enable: "true"
  druid.msq.intermediate.storage.type: s3
  druid.msq.intermediate.storage.bucket: [BUCKET NAME]
  druid.msq.intermediate.storage.prefix: /druidstorage
  druid.msq.intermediate.storage.tempDir: /opt/druid/var/tmp

To generate the initial table we ran the following script:

curl -L -H 'Content-Type: application/json' 
  "https://admin:######@router-service.druid:8080/druid/v2/sql/statements" -d '{
  "query": "[SQL]",
  "context":{
    "executionMode":"ASYNC",
    "selectDestination": "durableStorage"
  }
}' --user "admin:######"

Based on the query id in the response, the results of the query are paginated and accessible via the following endpoint:

curl -L -H 'Content-Type: application/json'  
  "https://admin:######@router-service.druid:8080/druid/v2/sql/statements/query-ID"  
  --user "admin:######" 

Conclusion

The live commentary bot we developed was the perfect blend of real-time analytics, historical context and predictive modelling. By harnessing the power of strokes-gained statistics, Apache Druid and 20 years of golf data we created a dynamic and engaging experience that elevated the way fans are able to interact with the tournament.

Watch the live commentary in text form at: https://x.com/openaigolf or https://x.com/ML_essi

And visit these blogs to see the next steps in producing the Live Golf Radio:

[embed]Repurposing real-time text through GPT Rephrasing text is straightforward just by asking the prompt. This is about fitting it into your real-time platformmedium.com

[embed]Using embedding vectors for semantic queries Suppose we want to make texts query accessible to a final user who lacks SQL experience. In that case, we will need a…medium.com