是否可以使用 Python 从 EventHub 中接收器停止的位置继续读取？答案

【问题标题】：Is it possible to continue reading from where receiver stopped in EventHub with Python?是否可以使用 Python 从 EventHub 中接收器停止的位置继续读取？
【发布时间】：2020-03-19 08:04:10
【问题描述】：

我正在使用官方 azure-eventhub 库从 EventHub 的主题中读取事件。我希望能够在接收器停止读取时，EventHub 保持接收器组的最后一个偏移量，这样当我再次开始读取时，我可以从停止的地方开始。那是在 Kafka 中使用提交策略完成的，但我在 azure-eventhub 库中找不到类似的东西。在 azure-eventhub 或任何其他用于 eventthub 的 python 库中是否有类似的东西？

【问题讨论】：

标签： python azure commit offset azure-eventhub

【解决方案1】：

您可以在 python 中使用 Event Processor Host 将检查点设置为 blob 存储。

详细的示例代码是 github 中的here，我们将其用于检查点目的。

在使用事件处理器主机之前，您应该拥有/创建一个azure storage account 用于保存检查点。和 create a container 在那个 azure 存储帐户中。示例代码中使用了storage account/account key/containerhere。

在示例中，这行代码context.checkpoint_async()用于设置检查点。

如果您对此有更多疑问，请告诉我。

【讨论】：

【解决方案2】：

您可以使用 azure-eventhub v5 来实现目标。 v5 sdk 以一种简单的方式集成了在 Azure 存储 Blob 中存储检查点的功能。

azure-eventhub v5 已于 2020 年 1 月 GAed，最新版本为 v5.2.0

它在 pypi 上可用：https://pypi.org/project/azure-eventhub/

请查看sample code 了解如何在 v5 中实现这一点：

#!/usr/bin/env python

# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

"""
An example to show receiving events from an Event Hub with checkpoint store doing checkpoint by batch.
In the `receive_batch` method of `EventHubConsumerClient`:
If no partition id is specified, the checkpoint_store are used for load-balance and checkpoint.
If partition id is specified, the checkpoint_store can only be used for checkpoint without load balancing.
"""

import os
import logging
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore

CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']
STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"]
BLOB_CONTAINER_NAME = "your-blob-container-name"  # Please make sure the blob container resource exists.

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)


def on_event_batch(partition_context, event_batch):
    log.info("Partition {}, Received count: {}".format(partition_context.partition_id, len(event_batch)))
    # put your code here
    partition_context.update_checkpoint()


def receive_batch():
    checkpoint_store = BlobCheckpointStore.from_connection_string(STORAGE_CONNECTION_STR, BLOB_CONTAINER_NAME)
    client = EventHubConsumerClient.from_connection_string(
        CONNECTION_STR,
        consumer_group="$Default",
        eventhub_name=EVENTHUB_NAME,
        checkpoint_store=checkpoint_store,
    )
    with client:
        client.receive_batch(
            on_event_batch=on_event_batch,
            max_batch_size=100,
            starting_position="-1",  # "-1" is from the beginning of the partition.
        )


if __name__ == '__main__':
    receive_batch()

我们还提供migration guide from v1 to v5 来迁移程序。

【讨论】：