Embarringly Parallel Structured Streaming: Eventhub and Databricks

Background

I work for a company that produces online games (1000's of them) and are looking for some guidance from the greater community. We have only been using Azure Databricks and Apache Spark for about a week, so chances are we doing some things wrong(This is why I'm reaching out)

Business Requirement

We are trying to use Structured Streaming and Azure Databricks to process around 17 Million Events using a sliding window. The business requirement states that we need to figure out how many players are playing a specific game for the last 24 hours, hopping forward every one minute. Essentially using the Sliding Window code window("EventTimeUtc", "24 hours", "1 minute")

EventHub Namespace Config

• Eventhubs (Topic): 1
• Incoming Events: 17 Million Events a Day
• Eventhub: 32 Partitions
• Throughput Units: 8
• Partitioning: Events are partitioned properly, Ie there is no relation between events in partition 1 and 2, so that can be processed in parallel.

Test A

With no aggregate query, ie just partitioning going into delta lake by MID, CID. Here is the stats.

• Input Rate 10K/sec
• Processing Rate: 10K/sec
• Time: 3-4 Seconds

From this, I'm satisfied that Event Hub is sending records at an appropriate speed (Might be a false assumption)

From there, I applied an aggregate query with approx_count_distinct and the rate drops significantly. I'm obviously expecting the rate to drop, but not so drastically.

• Input Rate 200 - 300/sec
• Processing Rate: 200 - 300/sec
• Time: 2-3 Min

Questions

1. What can I do to process the micro-batches in parallel. Ie. Align the amount of micro-batches triggered with the amount of partitions from Eventhub? Is it possible, Is it recommended?

2. Is there anything in my code that contributes to slowing down the processing rate, perhaps I'm not decoding and serializing properly?

Full Code

#######################################
# Configure Streaming From Event Hub  #
#######################################
# https://tsmatz.github.io/azure-databricks-exercise/exercise08-streaming-eventhub.html
# https://github.com/Azure/azure-event-hubs-spark/blob/master/docs/PySpark/structured-streaming-pyspark.md#user-configuration for Spark 3.1 (Encrypt Connection String Required)

# Library: com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.20
import json
from pyspark.sql.types import *
from pyspark.sql.functions import *
from datetime import datetime
from time import mktime

# Global Variables
conf = {}
startTime = "2021-07-15T00:00:00.000000Z"
endTime = "2021-07-15T23:59:59.000000Z"

# Set starting time eventhub stream
startingEventPosition = {
  "offset": None,
  "seqNo": -1, #not in use
  "enqueuedTime": startTime,
  "isInclusive": True
}

# Set end time eventhub stream
endingEventPosition = {
  "offset": None, #not in use
  "seqNo": -1, #not in use
  "enqueuedTime": endTime,
  "isInclusive": True
}

# Environment Variables
connectionString = "xxxxxx"
conf["eventhubs.connectionString"] = sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(connectionString)
conf["eventhubs.consumerGroup"] = "xxxxxx"
conf["eventhubs.startingPosition"] = json.dumps(startingEventPosition)
conf["eventhubs.endingPosition"] = json.dumps(endingEventPosition)

# Start Reading Event Hub Stream
streamingInputDF = (
  spark
    .readStream
    .format("eventhubs")
    .options(**conf)
    .load()
)
# Returns True for DataFrames that have streaming sources
streamingInputDF.isStreaming  

###############################
# Decode & Clean Stream Data  #
###############################

# Define Structure
event_schema = StructType([
  StructField("MID", IntegerType(), False),
  StructField("CID", IntegerType(), False),
  StructField("PID", IntegerType(), False),
  StructField("EventTimeUtc", TimestampType(), False),
  StructField("UserID", StringType(), False)
])

# Decode 
decodedInputDF = ( 
  streamingInputDF
    .select(from_json(col("body")
    .cast("string"), event_schema).alias("payload")
   )
)

# Clean and Flatten
cleanInputDF = (
  decodedInputDF
    .withColumn("EventTimeUtc",to_timestamp(col("payload.EventTimeUtc"), "yyyyMMdd HH:mm:ss"))
    .withColumn("MID",col("payload.MID"))
    .withColumn("CID",col("payload.CID"))
    .withColumn("UserID",col("payload.UserID"))
    .drop("payload")
)

# https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#reporting-metrics-using-dropwizard
spark.conf.set("spark.sql.streaming.metricsEnabled", "true")

#######################
# Sandbox Tests      ##
#######################

# https://databricks.com/blog/2016/05/19/approximate-algorithms-in-apache-spark-hyperloglog-and-quantiles.html
# https://docs.delta.io/latest/best-practices.html#language-python

# Test A - approx_count_distinct with window - query
test_a_query = (
  cleanInputDF
    .withWatermark("EventTimeUtc", "24 hours")
    .groupBy(
      "MID",
      "CID",
      window("EventTimeUtc", "24 hours", "1 minute").alias("HopWindow")
    ).agg(approx_count_distinct("UserID").alias("PlayerCount"))
)

# Test A - approx_count_distinct with window - output sink delta tables
test_a_output = (
  test_a_query
    .writeStream
    .format("delta")
    .outputMode("complete") # https://docs.databricks.com/delta/delta-streaming.html#complete-mode
    .option("checkpointLocation", "/mnt/delta/spinevents/_checkpoints/test_a")
    .queryName("test_a")
    .table("test_a") # https://docs.delta.io/latest/delta-streaming.html#delta-table-as-a-sink
)

Cluster Utilization

Monitoring Output