Reusing an Event Hub stream for multiple queries in Azure Data Bricks

Question

In Azure Databricks Structured Streaming (scala notebook, connected to Azure IoT Hub) I am opening a stream on the Event Hub compatible endpoint of Azure IoT Hub. Then I parse the incoming stream, based on the structured schema and I create 3 queries (groupBy) on the same stream. Most of the times (not always, it seems) I get an exception on one of the display queries around an epoch value on the partition. (see below) I am using a dedicated consumer group on which no other application is reading. So, I would guess that opening 1 stream and having multiple streaming queries against it would be supported?
Any suggestions, any explanations or ideas to solve this? (I would like to avoid having to create 3 consumer groups and defining the stream 3 times again)

Exception example:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 1064.0 failed 4 times, most recent failure: Lost task 3.3 in stage 1064.0 (TID 24790, 10.139.64.10, executor 7): java.util.concurrent.CompletionException: com.microsoft.azure.eventhubs.ReceiverDisconnectedException: New receiver with higher epoch of '0' is created hence current receiver with epoch '0' is getting disconnected. If you are recreating the receiver, make sure a higher epoch is used. TrackingId:xxxx, SystemTracker:iothub-name|databricks-db, Timestamp:2019-02-18T15:25:19, errorContext[NS: yyy, PATH: savanh-traffic-camera2/ConsumerGroups/databricks-db/Partitions/3, REFERENCE_ID: a0e445_7319_G2_1550503505013, PREFETCH_COUNT: 500, LINK_CREDIT: 500, PREFETCH_Q_LEN: 0]

This is my code: (cleaned up)

// Define schema and create incoming camera eventstream
val cameraEventSchema = new StructType()
    .add("TrajectId", StringType)
    .add("EventTime", StringType)
    .add("Country", StringType)
    .add("Make", StringType)

val iotHubParameters =
    EventHubsConf(cameraHubConnectionString)
    .setConsumerGroup("databricks-db")
    .setStartingPosition(EventPosition.fromEndOfStream)

val incomingStream = spark.readStream.format("eventhubs").options(iotHubParameters.toMap).load()

// Define parsing query selecting the required properties from the incoming telemetry data
val cameraMessages =
    incomingStream
    .withColumn("Offset", $"offset".cast(LongType))
    .withColumn("Time (readable)", $"enqueuedTime".cast(TimestampType))
    .withColumn("Timestamp", $"enqueuedTime".cast(LongType))
    .withColumn("Body", $"body".cast(StringType))
    // Select the event hub fields so we can work with them
    .select("Offset", "Time (readable)", "Timestamp", "Body")
    // Parse the "Body" column as a JSON Schema which we defined above
    .select(from_json($"Body", cameraEventSchema) as "cameraevents")
    // Now select the values from our JSON Structure and cast them manually to avoid problems
    .select(
        $"cameraevents.TrajectId".cast("string").alias("TrajectId"),
        $"cameraevents.EventTime".cast("timestamp").alias("EventTime"), 
        $"cameraevents.Country".cast("string").alias("Country"), 
        $"cameraevents.Make".cast("string").alias("Make") 
    )
    .withWatermark("EventTime", "10 seconds")

val groupedDataFrame = 
  cameraMessages 
      .groupBy(window($"EventTime", "5 seconds") as 'window)
      .agg(count("*") as 'count)
      .select($"window".getField("start") as 'window, $"count")
display(groupedDataFrame)

val makeDataFrame = 
  cameraMessages 
      .groupBy("Make")
      .agg(count("*") as 'count)
      .sort($"count".desc)

display(makeDataFrame)

val countryDataFrame = 
  cameraMessages 
      .groupBy("Country")
      .agg(count("*") as 'count)
      .sort($"count".desc)

display(countryDataFrame)

Vijay Kumar Sharma Vijay Kumar Sharma · Accepted Answer · 2019-02-19T07:48:45

You can store the stream data into a table or a file location, then you can run the multiple queries on that table or file, all are running in real time. For a file, you need to specify the schema while extracting the data into a data frame, so it's a good practice to write the stream data into a table.

cameraMessages.writeStream
 .format("delta") 
 .outputMode("append")
 .option("checkpointLocation","/data/events/_checkpoints/data_file")
 .table("events")

Now you can execute your queries on the table 'events'. And for the data frame -

cameraMessages = spark.readStream.table("events")

I have faced the same issue while using EventHub, and the above trick is work for me.

for using a file instead of table

//Write/Append streaming data to file
cameraMessages.writeStream
  .format("parquet")
  .outputMode("append")
  .option("checkpointLocation", "/FileStore/StreamCheckPoint.parquet")
  .option("path","/FileStore/StreamData")
  .start()
//Read data from the file, we need to specify the schema for it
val Schema = (
 new StructType()
    .add(StructField("TrajectId", StringType))
    .add(StructField("EventTime", TimestampType))
    .add(StructField("Country", StringType))
    .add(StructField("Make", StringType))
  )

  val cameraMessages = (
      sqlContext.readStream
                .option("maxEventsPerTrigger", 1)
                .schema(Schema)
                .parquet("/FileStore/StreamData")
         )

Reusing an Event Hub stream for multiple queries in Azure Data Bricks

1 Answers