Hi all,

I'm working with event hubs and data bricks to process and enrich data in real-time.

Doing a "simple" test, I'm getting some weird values (input rate vs processing rate) and I think I'm losing data:

If you can see, there is a peak with 5k records but it is never processed in the 5 minutes after.

The script that I'm using is:

 
conf = {}
 
conf['eventhubs.connectionString'] = sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(connectionString_bb_stream)
conf['eventhubs.consumerGroup'] = 'adb_jr_tesst'
conf['maxEventsPerTrigger'] = '350000'
conf['maxRatePerPartition'] = '350000'
conf['setStartingPosition'] = sc._jvm.org.apache.spark.eventhubs.EventPosition.fromEndOfStream
 
df = (spark.readStream 
                 .format("eventhubs") 
                 .options(**conf) 
                 .load()
            )
 
json_df = df.withColumn("body", from_json(col("body").cast('String'), jsonSchema))
Final_df = json_df.select(["sequenceNumber","offset", "enqueuedTime",col("body.*")])
Final_df = Final_df.withColumn("Key", sha2(concat(col('EquipmentId'), col('TagId'), col('Timestamp')), 256))
Final_df.display()

can you help me to understand why I'm "losing" data or how I can improve the process?

The cluster that I'm using is:

I think is a cluster configuration issue, but I'm not sure how to tackle that.

Thanks for the help, guys!