Microsoft Azure Event Hubs checkpointer implementation with Blob Storage Client Library for Python
Azure EventHubs Checkpoint Store client library for Python using Storage Blobs
Azure EventHubs Checkpoint Store is used for storing checkpoints while processing events from Azure Event Hubs.
This Checkpoint Store package works as a plug-in package to
EventHubConsumerClient. It uses Azure Storage Blob as the persistent store for maintaining checkpoints and partition ownership information.
Please note that this is a sync library, for async version of the Azure EventHubs Checkpoint Store client library, please refer to azure-eventhubs-checkpointstoreblob-aio.
Install the package
$ pip install --pre azure-eventhub-checkpointstoreblob
Python2.7, Python 3.5.3 or later.
Microsoft Azure Subscription: To use Azure services, including Azure Event Hubs, you'll need a subscription. If you do not have an existing Azure account, you may sign up for a free trial or use your MSDN subscriber benefits when you create an account.
Event Hubs namespace with an Event Hub: To interact with Azure Event Hubs, you'll also need to have a namespace and Event Hub available. If you are not familiar with creating Azure resources, you may wish to follow the step-by-step guide for creating an Event Hub using the Azure portal. There, you can also find detailed instructions for using the Azure CLI, Azure PowerShell, or Azure Resource Manager (ARM) templates to create an Event Hub.
Azure Storage Account: You'll need to have an Azure Storage Account and create a Azure Blob Storage Block Container to store the checkpoint data with blobs. You may follow the guide creating an Azure Block Blob Storage Account.
Checkpointing is a process by which readers mark or commit their position within a partition event sequence. Checkpointing is the responsibility of the consumer and occurs on a per-partition basis within a consumer group. This responsibility means that for each consumer group, each partition reader must keep track of its current position in the event stream, and can inform the service when it considers the data stream complete. If a reader disconnects from a partition, when it reconnects it begins reading at the checkpoint that was previously submitted by the last reader of that partition in that consumer group. When the reader connects, it passes the offset to the event hub to specify the location at which to start reading. In this way, you can use checkpointing to both mark events as "complete" by downstream applications, and to provide resiliency if a failover between readers running on different machines occurs. It is possible to return to older data by specifying a lower offset from this checkpointing process. Through this mechanism, checkpointing enables both failover resiliency and event stream replay.
Offsets & sequence numbers
Both offset & sequence number refer to the position of an event within a partition. You can think of them as a client-side cursor. The offset is a byte numbering of the event. The offset/sequence number enables an event consumer (reader) to specify a point in the event stream from which they want to begin reading events. You can specify a timestamp such that you receive events enqueued only after the given timestamp. Consumers are responsible for storing their own offset values outside of the Event Hubs service. Within a partition, each event includes an offset, sequence number and the timestamp of when it was enqueued.
- Create an Azure Storage Blobs
- Create an Azure EventHubs
- Consume events using a
Create an Azure Storage Blobs
The easiest way to create a
ContainerClient is to use a connection string.
from azure.storage.blob import ContainerClient container_client = ContainerClient.from_connection_string("my_storageacount_connection_string", "mycontainer")
For other ways of creating a
ContainerClient, go to Blob Storage library for more details.
The easiest way to create a
EventHubConsumerClient is to use a connection string.
from azure.eventhub import EventHubConsumerClient eventhub_client = EventHubConsumerClient.from_connection_string("my_eventhub_namespace_connection_string", event_hub_path="myeventhub")
For other ways of creating a
EventHubConsumerClient, refer to EventHubs library for more details.
Consume events using a
BlobPartitionManager to do checkpoint
from azure.eventhub import EventHubConsumerClient from azure.storage.blob import ContainerClient from azure.eventhub.extensions.checkpointstoreblob import BlobPartitionManager eventhub_connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>' storage_container_connection_str = '<< CONNECTION STRING OF THE STORAGE >>' storage_container_name = '<< STORAGE CONTAINER NAME>>' def do_operation(events): # do some operations to the events. pass def process_events(partition_context, events): do_operation(events) partition_context.update_checkpoint(events[-1]) def main(): storage_container_client = ContainerClient.from_connection_string(storage_container_connection_str, storage_container_name) partition_manager = BlobPartitionManager(storage_container_client) # use the BlobPartitonManager to save client = EventHubConsumerClient.from_connection_string(eventhub_connection_str, partition_manager=partition_manager, receive_timeout=5, retry_total=3) try: client.receive(process_events, "$default") except KeyboardInterrupt: client.close() if __name__ == '__main__': main()
Enabling logging will be helpful to do trouble shooting. Refer to Logging to enable loggers for related libraries.
- ./samples/event_processor_blob_storage_example.py - EventHubConsumerClient with blob partition manager example
Reference documentation is available here
azure.eventhub.extensions.checkpointstorebloblogger to collect traces from the library.
azure.eventhublogger to collect traces from the main azure-eventhub library.
azure.storage.bloblogger to collect traces from azure storage blob library.
uamqplogger to collect traces from the underlying uAMQP library.
- Enable AMQP frame level trace by setting
logging_enable=Truewhen creating the client.
If you encounter any bugs or have suggestions, please file an issue in the Issues section of the project.
This project welcomes contributions and suggestions. Most contributions require you to agree to a Contributor License Agreement (CLA) declaring that you have the right to, and actually do, grant us the rights to use your contribution. For details, visit https://cla.microsoft.com.
When you submit a pull request, a CLA-bot will automatically determine whether you need to provide a CLA and decorate the PR appropriately (e.g., label, comment). Simply follow the instructions provided by the bot. You will only need to do this once across all repos using our CLA.
BlobPartitionManagerthat uses Azure Blob Storage Block Blob to store EventProcessor checkpoint data
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
|Filename, size||File type||Python version||Upload date||Hashes|
|Filename, size azure_eventhub_checkpointstoreblob-1.0.0b5-py2.py3-none-any.whl (8.2 kB)||File type Wheel||Python version py2.py3||Upload date||Hashes View hashes|
|Filename, size azure-eventhub-checkpointstoreblob-1.0.0b5.zip (22.3 kB)||File type Source||Python version None||Upload date||Hashes View hashes|
Hashes for azure_eventhub_checkpointstoreblob-1.0.0b5-py2.py3-none-any.whl
Hashes for azure-eventhub-checkpointstoreblob-1.0.0b5.zip