Python 用Azure Event Hubs クライアントライブラリ - バージョン 5.11.5

[アーティクル]
11/13/2023

Azure Event Hubsは、1 秒あたり何百万ものイベントを取り込み、複数のコンシューマーにストリーミングできる、拡張性の高い発行サブスクライブサービスです。これにより、接続されたデバイスとアプリケーションによって生成される大量のデータを処理して分析できます。 Event Hubs がデータを収集したら、リアルタイム分析プロバイダーを使用するか、バッチ処理/ストレージアダプターを使用してデータを取得、変換、格納できます。 Azure Event Hubsの詳細については、「Event Hubs とは」を参照してください。

Azure Event Hubs クライアントライブラリにより、Azure Event Hubs イベントの発行と利用が可能になります。また、次の用途に使われます。

ビジネスインテリジェンスと診断の目的で、アプリケーションに関するテレメトリを生成します。
アプリケーションの状態に関する事実を公開します。これにより、関係のある当事者が、アクションを起こすトリガーとして監視し使うことができます。
ビジネスや他のエコシステム内で発生する興味深い操作と相互作用を観察し、疎結合されたシステムが密接に結びつくことなくやり取りできるようにします。
1 つ以上の発行元からイベントを受信し、エコシステムのニーズに合わせてそれらを変換し、変換されたイベントをコンシューマーが観察できるように新しいストリームに公開します。

作業の開始

前提条件

Python 3.7 以降。
Microsoft Azure サブスクリプション:Azure Event Hubsを含む Azure サービスを使用するには、サブスクリプションが必要です。既存の Azure アカウントをお持ちでない場合は、無料試用版にサインアップするか、アカウントの作成時に MSDN サブスクライバー特典を使用できます。
Event Hubs 名前空間とイベントハブ:Azure Event Hubsと対話するには、名前空間と Event Hub も使用できる必要があります。 Azure リソースの作成に慣れていない場合は、Azure portalを使用して Event Hub を作成するためのステップバイステップガイドに従ってください。ここでは、Azure CLI、Azure PowerShell、または Azure Resource Manager (ARM) テンプレートを使用してイベントハブを作成するための詳細な手順も確認できます。

パッケージをインストールする

pip を使用して Python 用Azure Event Hubs クライアントライブラリをインストールします。

$ pip install azure-eventhub

クライアントを認証する

Event Hubs との対話は、EventHubConsumerClient クラスまたは EventHubProducerClient クラスのインスタンスで始まります。クライアントオブジェクトをインスタンス化するには、ホスト名、SAS/AAD 資格情報、イベントハブ名、または接続文字列のいずれかが必要です。

接続文字列からクライアントを作成します。

Event Hubs クライアントライブラリがイベントハブと対話するには、Event Hubs 名前空間を作成するときに自動的に作成される接続文字列を使用するのが最も簡単な方法です。 Azure の共有アクセスポリシーに慣れていない場合は、ステップバイステップガイドに従って Event Hubs 接続文字列を取得できます。

メソッドはfrom_connection_string、フォームEndpoint=sb://<yournamespace>.servicebus.windows.net/;SharedAccessKeyName=<yoursharedaccesskeyname>;SharedAccessKey=<yoursharedaccesskey>とエンティティ名の接続文字列を Event Hub インスタンスに取り込みます。この接続文字列は、Azure portal から取得できます。

azure-identity ライブラリを使用してクライアントを作成します。

または、Credential オブジェクトを使用して、azure-identity パッケージを使用して AAD 経由で認証することもできます。

上記のリンク先のサンプルで示されているこのコンストラクターは、Event Hub インスタンスのホスト名とエンティティ名と、TokenCredential プロトコルを実装する資格情報を受け取ります。 azure-identity パッケージで使用できるプロトコルの実装TokenCredentialがあります。ホスト名の形式 <yournamespace.servicebus.windows.net>はです。
によって azure-identity提供される資格情報の種類を使用するには、パッケージをインストールしてください。 pip install azure-identity
さらに、非同期 API を使用するには、まず、などの aiohttp非同期トランスポートをインストールする必要があります。 pip install aiohttp
Azure Active Directory を使用する場合は、Azure Event Hubs データ所有者ロールなどの Event Hubs へのアクセスを許可するロールがプリンシパルに割り当てられている必要があります。 Event Hubs で Azure Active Directory 承認を使用する方法の詳細については、関連するドキュメントを参照してください。

主要な概念

EventHubProducerClient は、組み込みデバイスソリューション、モバイルデバイスアプリケーション、コンソールまたはその他のデバイスで実行されているゲームタイトル、一部のクライアントまたはサーバーベースのビジネスソリューション、または Web サイトの一部として、テレメトリデータ、診断情報、使用状況ログ、またはその他のログデータのソースです。
EventHubConsumerClient は、そのような情報を Event Hub から取得して処理します。処理には、集計、複雑な計算、フィルター処理が含まれる場合があります。生データまたは変換された形式で情報を配布または保存する処理が含まれる場合もあります。 Event Hub コンシューマーは、多くの場合、Azure Stream Analytics、Apache Spark、Apache Storm などの組み込みの分析機能を備えた、堅牢で大規模なプラットフォームインフラストラクチャパーツです。
パーティションは、Event Hub に保持される順序付けされた一連のイベントです。 Azure Event Hubs では、パーティション分割されたコンシューマーパターンを介してメッセージストリーミングを提供し、各コンシューマーがメッセージストリームの特定のサブセット (パーティション) のみを読み取ります。新しいイベントが到着すると、このシーケンスの末尾に追加されます。パーティションの数は、Event Hub の作成時に指定され、変更することはできません。
コンシューマーグループはは、Event Hub 全体のビューです。コンシューマーグループを使用すると、複数のコンシューマーアプリケーションが個別のイベントストリームビューを持つことができるようになり、それぞれの場所から独自のペースでストリームを個別に読み取ることができます。コンシューマーグループあたり最大 5 つのリーダーをパーティションに同時に設定できますが、特定のパーティションとコンシューマーグループの組み合わせには、アクティブな 1 つのコンシューマーのみをお勧めします。各アクティブリーダーでは、そのパーティションからすべてのイベントを受け取ります。同じパーティションに複数のリーダーがある場合は、重複するイベントを受信します。

その他の概念と詳細な説明については、「 Event Hubs の機能」を参照してください。また、AMQP の概念については、 OASIS Advanced Messaging Queuing Protocol (AMQP) バージョン 1.0 に詳しく記載されています。

スレッドセーフ

EventHubProducerClient または EventHubConsumerClient がスレッドセーフであることを保証するものではありません。スレッド間でこれらのインスタンスを再利用することはお勧めしません。これらのクラスをスレッドセーフな方法で使用するのは、実行中のアプリケーション次第です。

データモデル型はスレッド EventDataBatch セーフではありません。スレッド間で共有したり、クライアントメソッドと同時に使用したりしないでください。

例

次のセクションでは、次のような最も一般的な Event Hubs タスクをカバーするいくつかのコードスニペットを示します。

Event Hub を検査する
イベントを Event Hub に発行する
イベントハブからのイベントを使用する
イベントハブからのイベントをバッチで使用する
イベントをイベントハブに非同期的に発行する
イベントハブからのイベントを非同期的に使用する
イベントハブからのイベントを非同期的にバッチ処理で使用する
チェックポイントストアを使用してイベントを使用し、チェックポイントを保存する
EventHubConsumerClient を使用してIoT Hubを操作する

Event Hub を検査する

イベントハブのパーティション ID を取得します。

import os
from azure.eventhub import EventHubConsumerClient

CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']

consumer_client = EventHubConsumerClient.from_connection_string(
    conn_str=CONNECTION_STR,
    consumer_group='$Default',
    eventhub_name=EVENTHUB_NAME,
)

with consumer_client:
    pass # consumer_client is now ready to be used.

イベントを Event Hub に発行する

メソッドEventHubProducerClientをcreate_batch使用してオブジェクトをEventDataBatch作成し、メソッドをsend_batch使用して送信できます。バッチサイズの EventDataBatch 上限 (バイト単位) に達するまで、メソッドを使用して add イベントをに追加できます。

def send_event_data_batch(producer):
    # Without specifying partition_id or partition_key
    # the events will be distributed to available partitions via round-robin.
    event_data_batch = producer.create_batch()
    event_data_batch.add(EventData('Single message'))
    producer.send_batch(event_data_batch)

イベントハブからのイベントを使用する

EventHub からイベントを使用するには、複数の方法があります。イベントの受信時にコールバックをトリガーするだけの場合、 EventHubConsumerClient.receive メソッドは次のように使用されます。

import logging
from azure.eventhub import EventHubConsumerClient

connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)

logger = logging.getLogger("azure.eventhub")
logging.basicConfig(level=logging.INFO)

def on_event(partition_context, event):
    logger.info("Received event from partition {}".format(partition_context.partition_id))
    partition_context.update_checkpoint(event)

with client:
    client.receive(
        on_event=on_event,
        starting_position="-1",  # "-1" is from the beginning of the partition.
    )
    # receive events from specified partition:
    # client.receive(on_event=on_event, partition_id='0')

イベントハブからのイベントをバッチで使用する

上記のサンプルでは、受信時に各メッセージのコールバックをトリガーしますが、次のサンプルでは、イベントのバッチでコールバックをトリガーし、一度に数値を受信しようとします。

import logging
from azure.eventhub import EventHubConsumerClient

connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)

logger = logging.getLogger("azure.eventhub")
logging.basicConfig(level=logging.INFO)

def on_event_batch(partition_context, events):
    logger.info("Received event from partition {}".format(partition_context.partition_id))
    partition_context.update_checkpoint()

with client:
    client.receive_batch(
        on_event_batch=on_event_batch,
        starting_position="-1",  # "-1" is from the beginning of the partition.
    )
    # receive events from specified partition:
    # client.receive_batch(on_event_batch=on_event_batch, partition_id='0')

イベントをイベントハブに非同期的に発行する

メソッドEventHubProducerをcreate_batch使用してオブジェクトをEventDataBatch作成し、メソッドをsend_batch使用して送信できます。バッチサイズの EventDataBatch 上限 (バイト単位) に達するまで、メソッドを使用して add イベントをに追加できます。

import asyncio
from azure.eventhub.aio import EventHubProducerClient  # The package name suffixed with ".aio" for async
from azure.eventhub import EventData

connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

async def create_batch(client):
    event_data_batch = await client.create_batch()
    can_add = True
    while can_add:
        try:
            event_data_batch.add(EventData('Message inside EventBatchData'))
        except ValueError:
            can_add = False  # EventDataBatch object reaches max_size.
    return event_data_batch

async def send():
    client = EventHubProducerClient.from_connection_string(connection_str, eventhub_name=eventhub_name)
    batch_data = await create_batch(client)
    async with client:
        await client.send_batch(batch_data)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(send())

イベントハブからのイベントを非同期的に使用する

この SDK では、同期ベースのコードと asyncio ベースのコードの両方がサポートされています。上記のサンプルで示されているようにを受け取るが、aio 内では、次のものが必要です。

import logging
import asyncio
from azure.eventhub.aio import EventHubConsumerClient

connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

logger = logging.getLogger("azure.eventhub")
logging.basicConfig(level=logging.INFO)

async def on_event(partition_context, event):
    logger.info("Received event from partition {}".format(partition_context.partition_id))
    await partition_context.update_checkpoint(event)

async def receive():
    client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)
    async with client:
        await client.receive(
            on_event=on_event,
            starting_position="-1",  # "-1" is from the beginning of the partition.
        )
        # receive events from specified partition:
        # await client.receive(on_event=on_event, partition_id='0')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(receive())

イベントハブからのイベントを非同期的にバッチ処理で使用する

すべての同期関数は、aio でもサポートされています。上記の同期バッチ受信で示したように、asyncio 内で次のように同じ処理を実行できます。

import logging
import asyncio
from azure.eventhub.aio import EventHubConsumerClient

connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

logger = logging.getLogger("azure.eventhub")
logging.basicConfig(level=logging.INFO)

async def on_event_batch(partition_context, events):
    logger.info("Received event from partition {}".format(partition_context.partition_id))
    await partition_context.update_checkpoint()

async def receive_batch():
    client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)
    async with client:
        await client.receive_batch(
            on_event_batch=on_event_batch,
            starting_position="-1",  # "-1" is from the beginning of the partition.
        )
        # receive events from specified partition:
        # await client.receive_batch(on_event_batch=on_event_batch, partition_id='0')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(receive_batch())

チェックポイントストアを使用してイベントを使用し、チェックポイントを保存する

EventHubConsumerClient は、複数のパーティションから一度にイベントを受信し、同じイベントハブとコンシューマーグループを使用して他のコンシューマーとの負荷分散を可能にする高レベルのコンストラクトです。

これにより、ユーザーはチェックポイントを使用してイベントが処理されたときの進行状況を追跡することもできます。

チェックポイントは、Event Hub インスタンス内のコンシューマーグループの特定のパーティションからユーザーが最後に正常に処理したイベントを表します。では EventHubConsumerClient 、の CheckpointStore インスタンスを使用してチェックポイントを更新し、負荷分散アルゴリズムに必要な関連情報を格納します。

pypi にプレフィックス azure-eventhub-checkpointstore を付けて検索し、これをサポートするパッケージを検索し、そのようなパッケージの実装を CheckpointStore 使用します。同期ライブラリと非同期ライブラリの両方が用意されていることに注意してください。

次の例では、の EventHubConsumerClient インスタンスを作成し、を使用します BlobCheckpointStore。コードを実行するには、Azure Storage アカウントと BLOB コンテナーを作成する必要があります。

Azure Blob Storage チェックポイントストアの非同期およびAzure Blob Storage チェックポイントストア同期は、永続的なストアとしてAzure Blob Storage適用される実装の 1 つですCheckpointStore。

import asyncio

from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
storage_connection_str = '<< CONNECTION STRING FOR THE STORAGE >>'
container_name = '<<NAME OF THE BLOB CONTAINER>>'

async def on_event(partition_context, event):
    # do something
    await partition_context.update_checkpoint(event)  # Or update_checkpoint every N events for better performance.

async def receive(client):
    await client.receive(
        on_event=on_event,
        starting_position="-1",  # "-1" is from the beginning of the partition.
    )

async def main():
    checkpoint_store = BlobCheckpointStore.from_connection_string(storage_connection_str, container_name)
    client = EventHubConsumerClient.from_connection_string(
        connection_str,
        consumer_group,
        eventhub_name=eventhub_name,
        checkpoint_store=checkpoint_store,  # For load balancing and checkpoint. Leave None for no load balancing
    )
    async with client:
        await receive(client)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

EventHubConsumerClient を使用してIoT Hubを操作する

を使用EventHubConsumerClientして、IoT Hubを操作することもできます。これは、リンクされた EventHub からIoT Hubのテレメトリデータを受信する場合に便利です。関連付けられている接続文字列には送信要求がないため、イベントを送信することはできません。

接続文字列は、Event Hub と互換性のあるエンドポイント用である必要があることに注意してください。たとえば、"Endpoint=sb://my-iothub-namespace-[uid].servicebus.windows.net/;SharedAccessKeyName=my-SA-name;SharedAccessKey=my-SA-key;EntityPath=my-iot-hub-name"

Event Hubs 互換エンドポイントを取得するには、次の 2 つの方法があります。

Azure Portal でIoT Hubの "組み込みエンドポイント" を手動で取得し、そこから受信します。

from azure.eventhub import EventHubConsumerClient

connection_str = 'Endpoint=sb://my-iothub-namespace-[uid].servicebus.windows.net/;SharedAccessKeyName=my-SA-name;SharedAccessKey=my-SA-key;EntityPath=my-iot-hub-name'
consumer_group = '<< CONSUMER GROUP >>'
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group)

partition_ids = client.get_partition_ids()

組み込みの Event Hubs 互換エンドポイントをプログラムで取得します。「IoT Hub接続文字列のサンプル」を参照してください。

トラブルシューティング

さまざまな障害シナリオをazure-eventhubs診断する方法の詳細については、トラブルシューティングガイドを参照してください。

次のステップ

その他のサンプルコード

このライブラリを使用して Event Hubs との間でイベントを送受信する方法の詳細な例については、サンプルディレクトリを参照してください。

ドキュメント

リファレンスドキュメントについては、こちらを参照してください。

スキーマレジストリと Avro エンコーダー

EventHubs SDK は、スキーマレジストリサービスと Avro とうまく統合されています。詳細については、「スキーマレジストリ SDK 」と「スキーマレジストリ Avro Encoder SDK」を参照してください。

純粋な Python AMQP トランスポートと下位互換性のサポート

Azure Event Hubs クライアントライブラリは、純粋な Python AMQP 実装に基づいています。 uAMQP が必要な依存関係として削除されました。

基になるトランスポートとしてを使用 uAMQP するには:

pip を使用してインストール uamqp します。

$ pip install uamqp

クライアントの構築中にを渡します uamqp_transport=True 。

from azure.eventhub import EventHubProducerClient, EventHubConsumerClient

connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

client = EventHubProducerClient.from_connection_string(
    connection_str, eventhub_name=eventhub_name, uamqp_transport=True
)
client = EventHubConsumerClient.from_connection_string(
    connection_str, consumer_group, eventhub_name=eventhub_name, uamqp_transport=True
)

注: message 以前にを公開していた uamqp.Messageの EventData/EventDataBatch属性は非推奨になりました。によって EventData.message/EventDataBatch.message 返される "レガシ" オブジェクトは、移行を容易にするために導入されています。

ソースからの uAMQP ホイールのビルド

uAMQP がの基になる AMQP プロトコル実装azure-eventhubとして使用されることを意図している場合は、ほとんどの主要なオペレーティングシステムで uAMQP ホイールが見つかります。

を使用 uAMQP する予定で、uAMQP ホイールが提供されていないプラットフォームで実行している場合は、 uAMQP インストールガイダンスに従ってソースからインストールしてください。

フィードバックの提供

バグが発生した場合、または提案がある場合は、プロジェクトの [問題 ] セクションに問題を提出してください。

共同作成

このプロジェクトでは、共同作成と提案を歓迎しています。ほとんどの共同作成では、共同作成者使用許諾契約書 (CLA) にご同意いただき、ご自身の共同作成内容を使用する権利を Microsoft に供与する権利をお持ちであり、かつ実際に供与することを宣言していただく必要があります。詳細については、 https://cla.microsoft.com を参照してください。

pull request を送信すると、CLA を提供して PR (ラベル、コメントなど) を適宜装飾する必要があるかどうかを CLA ボットが自動的に決定します。ボットによって提供される手順にそのまま従ってください。この操作は、Microsoft の CLA を使用するすべてのリポジトリについて、1 回だけ行う必要があります。

このプロジェクトでは、Microsoft オープンソースの倫理規定を採用しています。詳しくは、「Code of Conduct FAQ (倫理規定についてよくある質問)」を参照するか、opencode@microsoft.com 宛てに質問またはコメントをお送りください。

インプレッション数

次の方法で共有

Python 用Azure Event Hubs クライアントライブラリ - バージョン 5.11.5

作業の開始

前提条件

パッケージをインストールする

クライアントを認証する

主要な概念

スレッドセーフ

例

Event Hub を検査する

イベントを Event Hub に発行する

イベントハブからのイベントを使用する

イベントハブからのイベントをバッチで使用する

イベントをイベントハブに非同期的に発行する

イベントハブからのイベントを非同期的に使用する

イベントハブからのイベントを非同期的にバッチ処理で使用する

チェックポイントストアを使用してイベントを使用し、チェックポイントを保存する

EventHubConsumerClient を使用してIoT Hubを操作する

トラブルシューティング

次のステップ

その他のサンプルコード

ドキュメント

スキーマレジストリと Avro エンコーダー

純粋な Python AMQP トランスポートと下位互換性のサポート

ソースからの uAMQP ホイールのビルド

フィードバックの提供

共同作成

フィードバック

その他のリソース

次の方法で共有

Python 用Azure Event Hubs クライアント ライブラリ - バージョン 5.11.5

作業の開始

前提条件

パッケージをインストールする

クライアントを認証する

主要な概念

スレッド セーフ

例

Event Hub を検査する

イベントを Event Hub に発行する

イベント ハブからのイベントを使用する

イベント ハブからのイベントをバッチで使用する

イベント をイベント ハブに非同期的に発行する

イベント ハブからのイベントを非同期的に使用する

イベント ハブからのイベントを非同期的にバッチ処理で使用する

チェックポイント ストアを使用してイベントを使用し、チェックポイントを保存する

EventHubConsumerClient を使用してIoT Hubを操作する

トラブルシューティング

次のステップ

その他のサンプル コード

ドキュメント

スキーマ レジストリと Avro エンコーダー

純粋な Python AMQP トランスポートと下位互換性のサポート

ソースからの uAMQP ホイールのビルド

フィードバックの提供

共同作成

フィードバック

その他のリソース

Python 用Azure Event Hubs クライアントライブラリ - バージョン 5.11.5

スレッドセーフ

イベントハブからのイベントを使用する

イベントハブからのイベントをバッチで使用する

イベントをイベントハブに非同期的に発行する

イベントハブからのイベントを非同期的に使用する

イベントハブからのイベントを非同期的にバッチ処理で使用する

チェックポイントストアを使用してイベントを使用し、チェックポイントを保存する

その他のサンプルコード

スキーマレジストリと Avro エンコーダー