Inconsistencies with Azure Event Hubs and Checkpoints

Paul Hernandez 631 Microsoft Employee

We are observing inconsistencies in the process of sending events to an event hub, reading events from a Databricks notebook using spark.readStream, and making use of checkpoint to read only the latest records coming to EventHub. While in normal cases, the count of events sent and received match, in some random cases, we are getting all of the events instead of just the increment, that is, the ones sent after the previous checkpoint. Could anyone please point to possible causes of this behavior?

BhargavaGunnam-MSFT 26,306 Reputation points Microsoft Employee

2024-01-25T21:46:25.7666667+00:00

Hello Paul Hernandez,

Can you please check if the checkpoint location is set correctly by verifying the path in the spark.readStream method?

In your query, can you check the initial position as mentioned in the below document?

https://docs.databricks.com/en/structured-streaming/delta-lake.html#complete-mode

ALso please review the logs/metrics for the streaming query to see if there are any errors or delays that could be causing the behavior you are observing.

Paul Hernandez 631 Microsoft Employee

Hi @BhargavaGunnam-MSFT , thanks for your input. We are setting the checkpoint not while writing but in the DataStreamWriter as specify in the screenshot, User's image

Here is a piece of the actual code:

df = spark \
  .readStream \
  .format("eventhubs") \
  .options(**ehConf) \
  .load()

bodyDF = df \
    .writeStream \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/checkpointstore") \
    .format("delta") \
    .start("/mnt/data/myevents")

We don't have a delta lake streaming source as specified in the link provided. Data goes from event hubs into spark streaming and in the real scenario, data is sink to an azure SQL db (in my snippet a delta table). I was also testing with a receiver locally in vscode. I noticed the local receiver creates a folder structure in the storage account using the namespace FQDN --> event hub name --> consumer group --> checkpoints. In the checkpoints directory one file per partition. The files are empty: User's image

I created and ran a script to retrieve the offsets by partition:

import asyncio
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
from azure.identity.aio import DefaultAzureCredential

BLOB_STORAGE_ACCOUNT_URL = "https://<------>.blob.core.windows.net"
BLOB_CONTAINER_NAME = "events"
EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "<------->.servicebus.windows.net"
EVENT_HUB_NAME = "validation_request"
#EVENT_HUB_NAME = "processed_records"

credential = DefaultAzureCredential()

async def main():

    # Create an Azure blob checkpoint store to store the checkpoints.
    checkpoint_store = BlobCheckpointStore(
        blob_account_url=BLOB_STORAGE_ACCOUNT_URL,
        container_name=BLOB_CONTAINER_NAME,
        credential=credential,
    )

    # Get the list of all checkpoints from the checkpoint store
    checkpoints = await checkpoint_store.list_checkpoints(EVENT_HUB_FULLY_QUALIFIED_NAMESPACE, EVENT_HUB_NAME, "$Default")

    # Now you can access the information you need:
    for checkpoint in checkpoints:
        print(f"Partition: {checkpoint['partition_id']}, Offset: {checkpoint['offset']}, ConsumerGroup: {checkpoint['consumer_group']}, EventHubName: {checkpoint['eventhub_name']}")

    # Close credential when no longer needed.
    await credential.close()
    
    await checkpoint_store.close()

if __name__ == "__main__":
    # Run the main method.
    asyncio.run(main())

The result is always the same, offsets do not get updated: User's image

Then I run my streaming job with the checkpoint configuration for the first time, using the same storage account and container for checkpointing, and a complete different structure was created. User's image

I would like to know how to programmatically get the current offsets per partition and application (if they are maintained with this granularity) Any help will be appreciated.

BhargavaGunnam-MSFT 26,306 Reputation points Microsoft Employee

2024-02-01T01:41:40.54+00:00

One possible reason I could think of is the checkpoint files are not being written to the Blob Storage container correctly. In your Spark Structured Streaming job, you are specifying the checkpoint location as /mnt/checkpointstore, which is a mount point for a Blob Storage container. However, in your code for retrieving the checkpoints, you are using a different Blob Storage container named events. This could be the reason why you are not seeing the expected offsets.

can you use the same Blob Storage container for both writing and reading the checkpoints and see if it helps resolve the issue.