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

最小権限のデータプレーンアクセス権を持つ承認済みのテナント／サブスクリプション ID と、別個のテスト用コンシューマーグループ／チェックポイントコンテナーを使用してください。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)

Event Hub の情報を取得する

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

使用する場合

このスキルは、概要に記載されたワークフローまたはアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記の範囲に明確に該当する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全性の境界、または成功基準が不足している場合は、作業を中断して確認してください。