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!