用于 python 的 Azure eventthub 库

Question

我正在使用 eventthub 来摄取大量事件。我有多个消费者正在运行一个扩展组，从具有多个分区的 eventthub 读取这些事件。我正在使用 python 中的 Azure SDK 并且对使用什么感到困惑。有eventthubconsumerclient、eventprocessorHost ....

我想使用一个库，我的多个消费者可以使用消费者组连接，分区动态分配给每个消费者，并在存储帐户中进行检查点，就像我使用 kafka 的方式一样。

Answer 1

更新：

对于生产使用，我建议您应该使用稳定版的事件中心 sdk。可以使用eph，示例代码为here。

---

我可以使用pre-release eventhub 5.0.0b6 来使用消费者组以及设置检查点。

但奇怪的是，在 blob 存储中，我可以看到为 eventthub 创建的 2 个文件夹：checkpoint 和 ownership 文件夹。在文件夹内，为分区创建了 blob，但 blob 为空。更奇怪的是，即使blob是空的，每次从eventthub读取时，它总是读取最新的数据（意味着它从不读取已经在同一个消费者组中读取的数据）。

您需要安装azure-eventhub 5.0.0b6 并使用pip install --pre azure-eventhub-checkpointstoreblob 安装azure-eventhub-checkpointstoreblob。对于 blob 存储，您应该安装最新的 version 12.1.0 of azure-storage-blob。

我关注这个sample。在此示例中，它使用 事件中心级别 连接字符串（不是 事件中心命名空间级别 连接字符串）。您需要通过导航创建一个 事件中心级别 连接字符串到 azure 门户 -> 你的 eventhub 命名空间 -> 你的事件中心实例 -> 共享访问策略 -> 单击“添加” -> 然后指定一个策略名称，然后选择权限。如果只想接收数据，只能选择监听权限。截图如下：

创建策略后，您可以按照以下屏幕截图复制连接字符串：

然后你可以按照下面的代码：

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

CONNECTION_STR = 'Endpoint=sb://ivanehubns.servicebus.windows.net/;SharedAccessKeyName=saspolicy;SharedAccessKey=xxx;EntityPath=myeventhub'
STORAGE_CONNECTION_STR = 'DefaultEndpointsProtocol=https;AccountName=xx;AccountKey=xxx;EndpointSuffix=core.windows.net'


def on_event(partition_context, event):
    # do something with event
    print(event)
    print('on event')
    partition_context.update_checkpoint(event)


if __name__ == '__main__':

    #the "a22" is the blob container name
    checkpoint_store = BlobCheckpointStore.from_connection_string(STORAGE_CONNECTION_STR, "a22")

    #the "$default" is the consumer group
    client = EventHubConsumerClient.from_connection_string(
        CONNECTION_STR, "$default", checkpoint_store=checkpoint_store)

    try:
        print('ok')
        client.receive(on_event)
    except KeyboardInterrupt:
        client.close()

测试结果：

Answer 2

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

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

请关注migration guide from v1 to v5 迁移您的程序。

使用checkpoint收货请关注sample code:

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()

还有一点值得注意的是，在 V5 中，我们使用 blob 的元数据来存储检查点和所有权信息，而不是在 v1 中将它们存储为 blob 的内容。所以在使用 v5 sdk 时，预计 blob 的内容为空。