Azure Databricks only gets Event Hub Data sent while its runnng

user_249868 0
Viewed 359

I'm trying to read Azure Event Hub data using databricks.

I have a producer running in nodejs, as well as a consumer for testing (on a different consumer group) and all seems to be running fine.

I am using the following pyspark code in databricks to get the data:

# Initialize event hub config dictionary with connectionString
connectionString = "Endpoint=sb://XXX.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=XXX;EntityPath=XXX"
ehConf = {}
ehConf['eventhubs.connectionString'] = sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(connectionString)
 
# Add consumer group to the ehConf dictionary
ehConf['eventhubs.consumerGroup'] = "databricks"
 
# Read events from the Event Hub
messages = spark.readStream.format("eventhubs").options(**ehConf).load()
 
# Visualize the Dataframe in realtime
display(messages)

The issue is that it only reads data from the stream if it is sent while the notebook is running. If i produce data and then run the notebook, it does not appear.

What am I missing? I want to use this to collect data from the stream every hour or so and save it.

Config: Databricks Runtime: 7.3LTS (Spark 3.0.1, Scala 2.12) Azure eventhub library: com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.21

azure-eventhub pyspark azure-databricks
1 Answers

You have two problems:

  1. display by default is using a temporary checkpoint, so when you run it next time, it don't know where to continue, so it starts again & again. If you continue to use display, add the `checkpointLocation="some_path" to the display call (see docs)
  2. by default, EventHubs connector reads only new data (that's why it consumes data only when running) - if you want to consume data from beginning (only on the initial call) - you need to add option eventhubs.startingPositions that encodes starting positions (see doc) - to start reading from beginning of the topic, assign following to this option:
import json

startingEventPosition = {
  "offset": -1,  
  "seqNo": -1,            #not in use
  "enqueuedTime": None,   #not in use
  "isInclusive": True
}

ehConf["eventhubs.startingPosition"] = json.dumps(startingEventPosition)
Alex_Ott_186270
Related