azure-eventhub-py

捆绑的上游支持文件

支持文件位于相对于技能的路径中。有关其固定来源、许可证、哈希值和导入变更，请参阅 UPSTREAM-NOTICE.md。

已恢复辅助脚本的安全性

scripts/setup_consumer.py 会向配置的 Event Hub 进行身份验证。send 会发布事件数据；receive 会打印事件正文和属性，并且可以写入或推进 Blob 检查点。实际使用前，请确认确切的命名空间、事件中心、负载／数量、起始位置、数据处理方式和检查点目标位置。

Azure Event Hubs SDK for Python

用于高吞吐量事件摄取的大数据流平台。

安装

pip install azure-eventhub azure-identity
# For checkpointing with blob storage
pip install azure-eventhub-checkpointstoreblob
# For the async examples instead:
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
)

客户端类型

客户端	用途
EventHubProducerClient	向 Event Hub 发送事件
EventHubConsumerClient	从 Event Hub 接收事件
BlobCheckpointStore	跟踪使用者进度

发送事件

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

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

with producer:
    # Create batch (handles size limits)
    event_data_batch = producer.create_batch()
    
    for i in range(10):
        try:
            event_data_batch.add(EventData(f"Event {i}"))
        except ValueError:
            # Batch is full, send and create new one
            producer.send_batch(event_data_batch)
            event_data_batch = producer.create_batch()
            event_data_batch.add(EventData(f"Event {i}"))
    
    # Send remaining
    producer.send_batch(event_data_batch)

发送到特定分区

# By partition ID
event_data_batch = producer.create_batch(partition_id="0")

# By partition key (consistent hashing)
event_data_batch = producer.create_batch(partition_key="user-123")

接收事件

简单接收

from azure.eventhub import EventHubConsumerClient

def on_event(partition_context, event):
    print(f"Partition: {partition_context.partition_id}")
    print(f"Data: {event.body_as_str()}")
    partition_context.update_checkpoint(event)

consumer = EventHubConsumerClient(
    fully_qualified_namespace="<namespace>.servicebus.windows.net",
    eventhub_name="my-eventhub",
    consumer_group="$Default",
    credential=DefaultAzureCredential()
)

with consumer:
    consumer.receive(
        on_event=on_event,
        starting_position="-1",  # Beginning of stream
    )

使用 Blob 检查点存储（示意）

from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
from azure.identity import DefaultAzureCredential

checkpoint_store = BlobCheckpointStore(
    blob_account_url="https://<account>.blob.core.windows.net",
    container_name="checkpoints",
    credential=DefaultAzureCredential()
)

consumer = EventHubConsumerClient(
    fully_qualified_namespace="<namespace>.servicebus.windows.net",
    eventhub_name="my-eventhub",
    consumer_group="$Default",
    credential=DefaultAzureCredential(),
    checkpoint_store=checkpoint_store
)

def on_event(partition_context, event):
    print(f"Received: {event.body_as_str()}")
    # Checkpoint after processing
    partition_context.update_checkpoint(event)

with consumer:
    consumer.receive(on_event=on_event)

异步客户端

from azure.eventhub import EventData
from azure.eventhub.aio import EventHubProducerClient, EventHubConsumerClient
from azure.identity.aio import DefaultAzureCredential
import asyncio

async def send_events():
    credential = DefaultAzureCredential()
    
    async with EventHubProducerClient(
        fully_qualified_namespace="<namespace>.servicebus.windows.net",
        eventhub_name="my-eventhub",
        credential=credential
    ) as producer:
        batch = await producer.create_batch()
        batch.add(EventData("Async event"))
        await producer.send_batch(batch)

async def receive_events():
    async def on_event(partition_context, event):
        print(event.body_as_str())
        await partition_context.update_checkpoint(event)
    
    async with EventHubConsumerClient(
        fully_qualified_namespace="<namespace>.servicebus.windows.net",
        eventhub_name="my-eventhub",
        consumer_group="$Default",
        credential=DefaultAzureCredential()
    ) as consumer:
        await consumer.receive(on_event=on_event)

asyncio.run(send_events())

事件属性

event = EventData("My event body")

# Set properties
event.properties = {"custom_property": "value"}
event.content_type = "application/json"

# Read properties (on receive)
print(event.body_as_str())
print(event.sequence_number)
print(event.offset)
print(event.enqueued_time)
print(event.partition_key)

获取事件中心信息

with producer:
    info = producer.get_eventhub_properties()
    print(f"Name: {info['name']}")
    print(f"Partitions: {info['partition_ids']}")
    
    for partition_id in info['partition_ids']:
        partition_info = producer.get_partition_properties(partition_id)
        print(f"Partition {partition_id}: {partition_info['last_enqueued_sequence_number']}")

最佳实践

1. 使用批次发送多个事件
2. 在生产环境中使用检查点存储，以确保可靠处理
3. 在高吞吐量场景中使用异步客户端
4. 使用分区键，确保在分区内按顺序传递
5. 处理批次大小限制——批次已满时捕获 ValueError
6. 使用上下文管理器（with／async with）正确清理资源
7. 为不同应用程序设置合适的使用者组

参考文件

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

适用场景

当要执行概述中描述的工作流或操作时，可使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 请勿将输出视为环境特定验证、测试或专家审查的替代品。
• 如果缺少所需输入、权限、安全边界或成功标准，请停止并请求澄清。