Azure Event Hubs データ接続

[アーティクル]
06/04/2024

Azure Event Hubs は、ビッグデータのストリーミングプラットフォームとなるイベントインジェストサービスです。 Azure Data Explorer は、お客様が管理する Event Hubs からの継続的なインジェストを提供します。

Event Hubs のインジェストパイプラインは、いくつかのステップで、Azure Data Explorer にイベントを転送します。まず、Azure portal でイベントハブを作成します。次に、Azure Data Explorer でターゲットテーブルを作成し、指定されたingestion プロパティを使用してデータを取り込みます。 Event Hubs 接続では、 events ルーティングに注意する必要があります。データは、イベントシステムプロパティのマッピングに従って、選択したプロパティを使用して埋め込むことができます。 Event Hubs への接続を作成して、イベントハブを作成し、イベントを送信します。このプロセスは、Azure portal、C# または Python によるプログラム、または Azure Resource Manager テンプレートを使用して管理できます。

Azure Data Explorer でのデータインジェストに関する一般的な情報については、「Azure Data Explorer のデータインジェスト概要」を参照してください。

Azure Data Explorer のデータ接続認証オプション

マネージド ID ベースのデータ接続 (推奨): マネージド ID ベースのデータ接続を使用することは、データソースに接続するための最も安全な方法です。データソースからデータをフェッチする機能を完全に制御できます。マネージド ID を使用したデータ接続のセットアップには、次の手順が必要です。
1. クラスターにマネージド ID を追加します。
2. データソースのマネージド ID にアクセス許可を付与します。 Azure Event Hubs からデータをフェッチするには、マネージド ID に Azure Event Hubs データ受信者のアクセス許可が必要です。
3. ターゲットデータベースで、マネージド ID ポリシーを設定します。
4. マネージド ID 認証を使用してデータ接続を作成し、データをフェッチします。
注意事項

マネージド ID のアクセス許可がデータソースから削除されると、データ接続は機能しなくなり、そのデータソースからデータをフェッチできなくなります。

キーベースのデータ接続: データ接続に対しマネージド ID の認証が指定されていない場合、接続は既定として自動的にキーベースの認証に設定されます。キーベースの接続では、リソース接続文字列 (Azure Event Hubs 接続文字列など) を使用してデータがフェッチされます。 Azure Data Explorer は、指定されたリソースのリソース接続文字列を取得し、それを安全に保存します。その後、接続文字列を使用して、データソースからデータがフェッチされます。

注意事項

キーがローテーションされると、データ接続は機能しなくなり、データソースからデータをフェッチできなくなります。この問題を解決するには、データ接続を更新または再作成します。

データ形式

データは EventData オブジェクトの形式でイベントハブから読み取られます。
サポートされる形式を確認してください。

Note

Event Hubs からのインジェストでは、RAW 形式はサポートされていません。
Azure Event Hubs スキーマレジストリとスキーマレス Avro はサポートされていません。
gzip 圧縮アルゴリズムを使用してデータを圧縮できます。 Compression はインジェストプロパティを使って動的に指定するか、静的なデータ接続設定で指定することができます。
バイナリ形式 (Avro、ApacheAvro、Parquet、ORC、W3CLOGFILE) では、データ圧縮はサポートされていません。
バイナリ形式と圧縮データでは、カスタムエンコードと埋め込みのシステムプロパティはサポートされていません。
バイナリ形式 (Avro、ApacheAvro、Parquet、ORC、W3CLOGFILE) とインジェストマッピングを使用する場合、インジェストマッピング定義内のフィールドの順序は、テーブル内の対応する列の順序と一致する必要があります。

Event Hubs のプロパティ

Azure Data Explorer では、次の Event Hubs プロパティがサポートされています。

イベントを関連するテーブルにルーティングするのに役立つインジェストプロパティの閉じられたセット。
指定されたマッピングに基づいてデータに埋め込むことができる、イベントシステムプロパティの閉じられたセット。

Note

メタデータをイベントに関連付けるために使用される Event Hubs カスタムプロパティの取り込みはサポートされていません。カスタムプロパティを取り込む必要がある場合は、イベントデータの本文でそれらを送信します。詳細については「カスタムプロパティの取り込み」を参照してください。

インジェストのプロパティ

インジェストプロパティは、インジェストプロセス、データのルーティング先と処理方法を指示します。 EventData.Properties を使用して、イベントインジェストのインジェストプロパティを指定できます。以下のプロパティを設定できます。

Note

プロパティ名は大文字と小文字が区別されます。

プロパティ	説明
データベース	大文字と小文字が区別される、ターゲットデータベースの名前。既定では、データはデータ接続に関連付けられているターゲットデータベースに取り込まれます。既定のデータベースをオーバーライドし、別のデータベースにデータを送信するには、このプロパティを使用します。これを行うには、最初にマルチデータベース接続として接続を設定する必要があります。
テーブル	大文字と小文字が区別される、既存のターゲットテーブルの名前。 [`Data Connection`] ペインで設定された `Table` をオーバーライドします。
形式	データ形式。 [`Data Connection`] ペインで設定された `Data format` をオーバーライドします。
IngestionMappingReference	使用する既存のインジェストマッピングの名前。 [`Data Connection`] ペインで設定された `Column mapping` をオーバーライドします。
圧縮	データ圧縮、`None` (既定値)、または `gzip`。
[エンコード]	データエンコード (既定値は UTF8)。 .NET でサポートされているエンコードのいずれかを指定できます。
タグ	JSON 配列文字列として書式設定された、取り込まれたデータに関連付けられるタグの一覧。タグを使用すると、パフォーマンスに影響します。
RawHeaders	イベントソースが Kafka で、他のルーティングプロパティを読み取るには Azure Data Explorer でバイト配列の逆シリアル化を使用する必要があることを示します。値は無視されます。

Note

カスタム取得の開始日が指定されていない限り、データ接続の作成後にエンキューされたイベントのみが取り込まれます。いずれの場合も、ルックバック期間は実際のイベントハブのリテンション期間を超えることはできません。

イベントルーティング

クラスターへのデータ接続を作成するときに、取り込まれたデータの送信先のルーティングを指定できます。既定のルーティングは、ターゲットデータベースに関連付けられている接続文字列で指定されたターゲットテーブルです。データに対する既定のルーティングは、"静的ルーティング" と呼ばれることもあります。前の段落で説明した 1 つ以上のイベントデータプロパティを設定することで、データの代替ルーティングと処理のオプションを指定できます。

Note

Event Hubs データ接続は、Event Hub から読み取ったすべてのイベントの処理を試みます。また、何らかの理由で処理できないすべてのイベントは、インジェストエラーとして報告されます。 Azure Data Explorer のインジェストを監視する方法についてはこちらを参照してください。

イベントデータを別のデータベースにルーティングする

代替データベースへのデータのルーティングは、既定では無効になっています。データを別のデータベースに送信するには、最初に、マルチデータベース接続として接続を設定する必要があります。この機能は、Azure portal Azure ポータル、 C# または Python 管理 SDK、または ARM テンプレートを使用して有効にすることができます。データベースルーティングを許可するために使用されるユーザー、グループ、サービスプリンシパル、またはマネージド ID には、少なくともクラスターに対する共同作成者ロールと書き込みアクセス許可が必要です。

別のデータベースを指定するには、データベースのインジェストプロパティを設定します。

警告

マルチデータベースのデータ接続として接続を設定せずに別のデータベースを指定すると、インジェストが失敗します。

イベントデータを別のテーブルにルーティングする

各イベントに別のテーブルを指定するには、Table、Format、Compression、およびインジェストのプロパティのマッピングを設定します。 EventData.Properties の指定に従って、取り込まれたデータが接続で動的にルーティングされ、このイベントの静的プロパティがオーバーライドされます。

次の例では、イベントハブの詳細を設定し、気象メトリックデータを代替データベース (MetricsDB) とテーブル (WeatherMetrics) に送信する方法を示します。データは JSON 形式で、mapping1 はテーブル WeatherMetrics で事前定義されています。

// This sample uses Azure.Messaging.EventHubs which is a .Net Framework library.
await using var producerClient = new EventHubProducerClient("<eventHubConnectionString>");
// Create the event and add optional "dynamic routing" properties
var eventData = new EventData(Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(
    new { TimeStamp = DateTime.UtcNow, MetricName = "Temperature", Value = 32 }
)));
eventData.Properties.Add("Database", "MetricsDB");
eventData.Properties.Add("Table", "WeatherMetrics");
eventData.Properties.Add("Format", "json");
eventData.Properties.Add("IngestionMappingReference", "mapping1");
eventData.Properties.Add("Tags", "['myDataTag']");
var events = new[] { eventData };
// Send events
await producerClient.SendAsync(events);

Event Hubs システムプロパティのマッピング

システムプロパティは、イベントがエンキューされた時点で Event Hubs サービスによって設定されるフィールドです。 Azuer Data Explorer Event Hubs データ接続では、特定のマッピングに基づいて、テーブルに取り込まれたデータに、定義済みのシステムプロパティのセットを埋め込むことができます。

Note

システムプロパティの埋め込みは、json および表形式 (つまり JSON、MultiJSON、CSV、TSV、PSV、SCsv、SOHsv、TSVE) でサポートされています。
サポートされていない形式 (例: TXT や、Parquet、Avro などの圧縮形式) を使用してもデータは取り込まれますが、プロパティは無視されます。
イベントハブメッセージの圧縮が設定されている場合、システムプロパティの埋め込みはサポートされません。このようなシナリオでは、該当するエラーが生成され、データは取り込まれません。
表形式データの場合、システムプロパティは単一レコードのイベントメッセージでのみサポートされます。
json データの場合、システムプロパティは複数レコードのイベントメッセージでもサポートされます。このような場合、システムプロパティは、イベントメッセージの最初のレコードにのみ追加されます。
CSV マッピングの場合、プロパティは、データ接続の作成に記載されている順序でレコードの先頭に追加されます。将来変更される可能性があるため、これらのプロパティの順序に依存しないでください。
JSON マッピングの場合、システムプロパティの表のプロパティ名に従ってプロパティが追加されます。

Event Hubs サービスでは、次のシステムプロパティが公開されます。

プロパティ	データ型	説明
x-opt-enqueued-time	`datetime`	イベントがエンキューされた UTC 時刻
x-opt-sequence-number	`long`	イベントハブのパーティションストリーム内のイベントの論理シーケンス番号
x-opt-offset	`string`	イベントハブのパーティションストリームからのイベントのオフセット。このオフセット識別子は、イベントハブストリームのパーティション内で一意です
x-opt-publisher	`string`	発行元の名前 (発行元のエンドポイントにメッセージが送信された場合)
x-opt-partition-key	`string`	イベントが格納されている、対応するパーティションのパーティションキー

IoT Central イベントハブを操作するときに、IoT Hub のシステムプロパティーをペイロードに埋め込むこともできます。完全な一覧については、IoT Hub のシステムプロパティに関する記事を参照してください。

テーブルの [データソース] セクションで [イベントシステムのプロパティ] を選択した場合は、テーブルのスキーマとマッピングにプロパティを含める必要があります。

スキーママッピングの例

テーブルスキーママッピングの例

データに 3 つの列 (TimeStamp、MetricName、および Value) が含まれており、含めるプロパティが x-opt-enqueued-time および x-opt-offset の場合は、次のコマンドを使用してテーブルスキーマを作成または変更します。

    .create-merge table TestTable (TimeStamp: datetime, MetricName: string, Value: int, EventHubEnqueuedTime:datetime, EventHubOffset:string)

CSV マッピングの例

次のコマンドを実行して、レコードの先頭にデータを追加します。序数値に注意してください。

    .create table TestTable ingestion csv mapping "CsvMapping1"
    '['
    '   { "column" : "TimeStamp", "Properties":{"Ordinal":"2"}},'
    '   { "column" : "MetricName", "Properties":{"Ordinal":"3"}},'
    '   { "column" : "Value", "Properties":{"Ordinal":"4"}},'
    '   { "column" : "EventHubEnqueuedTime", "Properties":{"Ordinal":"0"}},'
    '   { "column" : "EventHubOffset", "Properties":{"Ordinal":"1"}}'
    ']'

JSON マッピングの例

データは、システムプロパティのマッピングを使用して追加されます。これらのコマンドを実行します。

    .create table TestTable ingestion json mapping "JsonMapping1"
    '['
    '    { "column" : "TimeStamp", "Properties":{"Path":"$.TimeStamp"}},'
    '    { "column" : "MetricName", "Properties":{"Path":"$.MetricName"}},'
    '    { "column" : "Value", "Properties":{"Path":"$.Value"}},'
    '    { "column" : "EventHubEnqueuedTime", "Properties":{"Path":"$.x-opt-enqueued-time"}},'
    '    { "column" : "EventHubOffset", "Properties":{"Path":"$.x-opt-offset"}}'
    ']'

Event Hubs Capture Avro ファイルのスキーママッピング

Event Hubs データを使う方法の 1 つは、Azure Blob Storage または Azure Data Lake Storage の Azure Event Hubs を通じてイベントをキャプチャすることです。その後、Azure Data Explorer の Event Grid データ接続を使って、書き込まれたキャプチャファイルを取り込むことができます。

キャプチャしたファイルのスキーマは、Event Hubs に送信された元のイベントのスキーマとは異なります。この違いを考慮した上で、宛先テーブルのスキーマを設計する必要があります。具体的には、イベントのペイロードはキャプチャファイルではバイト配列で表され、この配列は Event Grid の Azure Data Explorer データ接続では自動的にデコードされません。 Event Hubs Avro キャプチャデータのファイルスキーマの詳細については、「キャプチャされた Avro ファイルを Azure Event Hubs で展開する」を参照してください。

イベントペイロードを正しくデコードするには:

キャプチャしたイベントの Body フィールドを、宛先テーブルの型 dynamic の列にマップします。
unicode_codepoints_to_string() 関数を使ってバイト配列を読み取り可能な文字列に変換する更新ポリシーを適用します。

カスタムプロパティの取り込み

Event Hubs からイベントを取り込む場合、データはイベントデータオブジェクトの body セクションから取得されます。ただし、Event Hubs のカスタムプロパティは、オブジェクトの properties セクションで定義され、取り込まれません。顧客のプロパティを取り込むには、オブジェクトのbody セクションのデータに埋め込む必要があります。

次の例では、Event Hubs (左側) によって "定義された" カスタムプロパティ customProperty を含むイベントデータオブジェクトと、インジェスト (右側) に必要な "埋め込み" プロパティを比較しています。

{
"body":{
"value": 42
},
"properties":{
"customProperty": "123456789"
}
}

{
"body":{
"value": 42,
"customProperty": "123456789"
}
}

次のいずれかの方法を使用して、イベントデータオブジェクトの body セクション内のデータにカスタムプロパティを埋め込むことができます。

Event Hubs では、イベントデータオブジェクトを作成するときに、オブジェクトの body セクションのデータの一部としてカスタムプロパティを埋め込みます。
Azure Stream Analytics を使用して、イベントハブからのイベントを処理し、カスタムプロパティをイベントデータに埋め込みます。 Azure Stream Analytics から、Azure Data Explorer 出力コネクタを使用してデータをネイティブに取り込んだり、データを別のイベントハブにルーティングして、そこから自分のクラスターにルーティングしたりすることができます。
Azure Functions を使用して、カスタムプロパティを追加し、データを取り込みます。

イベントハブを作成する

まだ用意していない場合は、イベントハブを作成します。イベントハブへの接続は、Azure portal、C# または Python によるプログラム、または Azure Resource Manager テンプレートを使用して管理できます。

Note

イベントハブの作成後にパーティションを動的に追加する機能は、Event Hubs の Premium と Dedicated レベルでのみ使用できます。パーティション数を選択する場合は、長期間にわたるスケールを考慮してください。
コンシューマーグループは、コンシューマーごとに一意である "必要があります"。 Azure Data Explorer 接続専用のコンシューマーグループを作成します。

リージョン間 Event Hubs データ接続

パフォーマンスを最大限に高めるには、クラスターと同じリージョンにイベントハブを作成します。これができない場合は、 Premium または Dedicated Event Hubs レベルの使用を検討してください。レベルの比較については、「 Compare Azure Event Hubs レベルを参照してください。

送信イベント

データを生成してイベントハブに送信するサンプルアプリをご覧ください。

Note

Event Hubs から Azure Data Explorer へのイベントの効率的な処理を有効にするには、パーティション間でのイベントの不均衡な分散を回避します。マッピングが不均一な場合、の情報開示の待機時間が長くなる可能性があります。詳細については、「パーティションへのイベントのマッピング」を参照してください。

geo ディザスターリカバリーソリューションの設定

イベントハブには、geo ディザスターリカバリーソリューションが用意されています。 Azure Data Explorer では Alias イベントハブ名前空間がサポートされていません。ソリューションに geo ディザスターリカバリーを実装するには、2 つのイベントハブデータ接続を作成します。1 つはプライマリ名前空間用で、もう 1 つはセカンダリ名前空間用です。 Azure Data Explorer は、両方のイベントハブ接続をリッスンします。

Note

プライマリ名前空間からセカンダリ名前空間へのフェールオーバーを実装するのは、ユーザーの責任です。

次の方法で共有

Azure Event Hubs データ接続

Azure Data Explorer のデータ接続認証オプション

データ形式

Event Hubs のプロパティ

インジェストのプロパティ

イベントルーティング

イベントデータを別のデータベースにルーティングする

イベントデータを別のテーブルにルーティングする

Event Hubs システムプロパティのマッピング

スキーママッピングの例

Event Hubs Capture Avro ファイルのスキーママッピング

カスタムプロパティの取り込み

イベントハブを作成する

リージョン間 Event Hubs データ接続

送信イベント

geo ディザスターリカバリーソリューションの設定

フィードバック

その他のリソース

次の方法で共有

Azure Event Hubs データ接続

Azure Data Explorer のデータ接続認証オプション

データ形式

Event Hubs のプロパティ

インジェストのプロパティ

イベント ルーティング

イベント データを別のデータベースにルーティングする

イベント データを別のテーブルにルーティングする

Event Hubs システム プロパティのマッピング

スキーマ マッピングの例

Event Hubs Capture Avro ファイルのスキーマ マッピング

カスタム プロパティの取り込み

イベント ハブを作成する

リージョン間 Event Hubs データ接続

送信イベント

geo ディザスター リカバリー ソリューションの設定

関連するコンテンツ

フィードバック

その他のリソース

イベントルーティング

イベントデータを別のデータベースにルーティングする

イベントデータを別のテーブルにルーティングする

Event Hubs システムプロパティのマッピング

スキーママッピングの例

Event Hubs Capture Avro ファイルのスキーママッピング

カスタムプロパティの取り込み

イベントハブを作成する

geo ディザスターリカバリーソリューションの設定