SuperAPI AI 工具集
合作ENAI 行业资讯收藏
Skill 广场azure-eventhub-py

azure-eventhub-py

数据与分析

查看项目

更新于 10-16 01:57

技能介绍

用途

Azure Event Hubs SDK for Python 用于大数据流式处理平台的高吞吐量事件摄取，适用于生产者、消费者以及检查点（checkpointing）等流式场景。

主要能力

  • 客户端类型：EventHubProducerClient 用于向 Event Hub 发送事件；EventHubConsumerClient 用于从 Event Hub 接收事件；BlobCheckpointStore 用于跟踪消费者进度。
  • 发送事件：支持创建批次（自动处理大小限制），可在批次已满时捕获 ValueError 并发送新批次。可发送到指定分区，方式包括 partition_id 或 partition_key（一致性哈希）。
  • 接收事件：提供简单接收方式，通过回调函数处理事件，并在回调中调用 update_checkpoint 更新检查点。可设置 starting_position（如 "-1" 表示流的开头）。
  • 生产环境检查点：通过 BlobCheckpointStore 结合 Blob 存储实现可靠的进度跟踪，在事件处理后调用 partition_context.update_checkpoint(event)。
  • 异步客户端：提供 azure.eventhub.aio 下的异步 EventHubProducerClient 与 EventHubConsumerClient，适合高吞吐场景。异步场景下检查点更新为 await partition_context.update_checkpoint(event)。
  • 事件属性：可设置 properties、content_type 等；接收时可读取 body_as_str()、sequence_number、offset、enqueued_time、partition_key 等。
  • 获取 Event Hub 信息：通过 producer.get_eventhub_properties() 获取名称与分区列表，通过 get_partition_properties(partition_id) 获取分区详情（如 last_enqueued_sequence_number）。

安装方式

pip install azure-eventhub azure-identity
# For checkpointing with blob storage
pip install azure-eventhub-checkpointstoreblob-aio

环境变量

EVENT_HUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net
EVENT_HUB_NAME=my-eventhub
STORAGE_ACCOUNT_URL=https://<account>.blob.core.windows.net
CHECKPOINT_CONTAINER=checkpoints

认证

使用 DefaultAzureCredential 进行认证：

from azure.identity import DefaultAzureCredential
from azure.eventhub import EventHubProducerClient, EventHubConsumerClient

credential = DefaultAzureCredential()
namespace = "<namespace>.servicebus.windows.net"
eventhub_name = "my-eventhub"

# Producer
producer = EventHubProducerClient(
    fully_qualified_namespace=namespace,
    eventhub_name=eventhub_name,
    credential=credential
)

# Consumer
consumer = EventHubConsumerClient(
    fully_qualified_namespace=namespace,
    eventhub_name=eventhub_name,
    consumer_group="$Default",
    credential=credential
)

适用场景

  • 高吞吐量事件摄取与流式处理。
  • 实现事件生产者与消费者。
  • 需要分区内有序投递（使用 partition keys）。
  • 生产环境需要可靠的消费进度跟踪（使用检查点存储）。
  • 高吞吐场景（使用异步客户端）。

最佳实践

  • 发送多个事件时使用批次。
  • 生产环境使用检查点存储以实现可靠处理。
  • 高吞吐场景使用异步客户端。
  • 使用分区键实现分区内有序投递。
  • 处理批次大小限制——批次已满时捕获 ValueError。
  • 使用上下文管理器（with/async with）确保正确清理。
  • 为不同应用设置合适的消费者组。

参考文件

  • references/checkpointing.md：检查点存储模式、Blob 检查点、检查点策略。
  • references/partitions.md：分区管理、负载均衡、起始位置。
  • scripts/setup_consumer.py：用于 Event Hub 信息、消费者设置以及事件发送/接收的 CLI。

限制

  • 仅在任务明确匹配上述范围时使用本技能。
  • 不要将输出视为环境相关验证、测试或专家审查的替代品。
  • 如果缺少必需的输入、权限、安全边界或成功标准，请停止并请求澄清。