Azure Data Explorer の Event Hubs データ接続を作成する

Azure Data Explorer では、ビッグ データ ストリーミング プラットフォームとイベント インジェストのサービスである Event Hubs からのインジェストを行うことができます。 Event Hubs は、1 秒あたり数百万件のイベントをほぼリアルタイムで処理できます。

この記事では、イベント ハブに接続してデータを Azure Data Explorer に取り込みます。 Event Hubs からの取り込みの概要については、「Azure Event Hubs データ接続」を参照してください。

Kusto SDK を使用して接続を作成する方法については、「SDK を使用して Event Hubs データ接続を作成するを参照してください。

以前の SDK バージョンに基づくサンプル コードについては、アーカイブ記事を参照してください。

イベント ハブ データ接続を作成する

このセクションでは、イベント ハブと Azure Data Explorer テーブルとの間の接続を確立します。 この接続が存在する間は、データがイベント ハブからターゲットのテーブルに送信されます。 イベント ハブが別のリソースまたはサブスクリプションに移動した場合は、接続を更新または再作成する必要があります。

データの取得

前提条件

• Microsoft アカウントまたは Microsoft Entra ユーザー ID。 Azure サブスクリプションは不要です。
• Azure Data Explorer クラスターとデータベース。 クラスターとデータベースを作成します。
• ストリーミング インジェストは、Azure Data Explorer クラスターで 構成する必要があります。

データを取得する

1. 左側のメニューから Query を選択します。

2. データを取り込むデータベースを右クリックします。 データの取得を選択します。

   

ソース

[データの取り込み] ウィンドウで [ソース] タブが選択されます。

使用可能な一覧からデータ ソースを選択します。 この例では Event Hubs からデータを取り込みます。

構成

1. ターゲット データベースとテーブルを選択します。 新しいテーブルにデータを取り込む場合は、[+ 新しいテーブル] を選択し、テーブル名を入力します。

   Note

   テーブル名には、スペース、英数字、ハイフン、アンダースコアを含め、最大 1024 文字を使用できます。 特殊文字はサポートされていません。

   

2. 以下のフィールドを設定します。

   設定	フィールドの説明
   サブスクリプション	イベントハブ リソースが配置されているサブスクリプション ID。
   イベント ハブの名前空間	名前空間を識別する名前。
   イベント ハブ	必要なイベント ハブ
   コンシューマー グループ	イベントで定義されたコンシューマー グループ
   データ接続名	データ接続を識別する名前。
   高度なフィルター
   圧縮	イベント ハブ メッセージ ペイロードの圧縮の種類。
   イベント システム プロパティ	イベント ハブのシステム プロパティ。 イベント メッセージごとに複数のレコードがある場合、システム プロパティは最初のものに追加されます。 システム プロパティを追加する場合は、テーブル スキーマとマッピングを作成または更新して、選択したプロパティを含めます。
   イベント取得の開始日	データ接続は、 Event 取得の開始日後に作成された既存の Event Hubs イベントを取得します。 取得できるのは、Event Hubs のリテンション期間によって保持されるイベントのみです。 Event 取得の開始日が指定されていない場合、既定の時刻はデータ接続が作成される時刻です。

3. [次へ] を選択します

検査

[検査] タブが開き、データのプレビューが表示されます。

インジェスト プロセスを完了するには、[完了] を選択します。

必要に応じて、次の操作を行います。

• プレビュー ウィンドウに表示されるデータが完全でない場合は、必要なすべてのデータ フィールドを含むテーブルを作成するために、さらにデータが必要になることがあります。 次のコマンドを使用して、イベント ハブから新しいデータをフェッチします。

  • 新しいデータの破棄とフェッチ: 表示されたデータを破棄し、新しいイベントを検索します。

  • Fetch more data (さらにデータをフェッチする) :既に見つかったイベントに加えて、さらにイベントを検索します。

    注意

    データのプレビューを表示するには、イベント ハブがイベントを送信している必要があります。

• [コマンド ビューアー] を選択し、入力から生成される自動コマンドを表示してコピーします。

• [スキーマ定義ファイル] ドロップダウンを使用して、スキーマを推論する元のファイルを変更します。

• ドロップダウンから必要な形式を選択して、自動的に推論されるデータの形式を変更します。 インジェストについては、Azure Data Explorer でサポートされている Data 形式を参照してください。

• 列を編集します。

• データ型に基づく [詳細] オプションを確認します。

列の編集

Note

• 表形式 (CSV、TSV、PSV) では、列を 2 回マップすることはできません。 既存の列にマップするには、最初に新しい列を削除します。
• 既存の列の型を変更することはできません。 異なる形式の列にマップしようとすると、空の列になってしまう場合があります。

テーブルに加えることができる変更は、次のパラメーターによって異なります。

• テーブルの種類が新規かまたは既存か
• マッピングの種類が新規かまたは既存か

テーブルの種類です。	マッピングの種類	使用可能な調整
新しいテーブル	新しいマッピング	列の名前変更、データ型の変更、データ ソースの変更、マッピング変換、列の追加、列の削除
既存のテーブル	新しいマッピング	新しい列の追加 (その後、データ型の変更、名前変更、更新が可能)
既存のテーブル	既存のマッピング	なし

マッピング変換

一部のデータ形式マッピング (Parquet、JSON、Avro) では、簡単な取り込み時の変換がサポートされています。 マッピング変換を適用するには、[列の編集] ウィンドウで列を作成または更新します。

マッピング変換は、データ型が int または long であるソースを使用して、string または datetime 型の列に対して実行できます。 サポートされているマッピング変換は次のとおりです。

• DateTimeFromUnixSeconds
• DateTimeFromUnixMilliseconds
• DateTimeFromUnixMicroseconds
• DateTimeFromUnixNanoseconds

データ型に基づく [詳細] オプション

表形式 (CSV、TSV、PSV):

• テーブル形式を 既存のテーブルに取り込む場合、 Advanced>Keep 現在のテーブル スキーマを選択できます。 表形式データには、ソース データを既存の列にマップするために使用される列名が必ずしも含まれるとは限りません。 このオプションをオンにすると、マッピングは順番に行われ、テーブル スキーマは同じままになります。 このオプションをオフにすると、データ構造に関係なく、受信するデータに対して新しい列が作成されます。

• 最初の行を列名として使用するには、[詳細]>[最初の行を列ヘッダーにする] を選択します。

  

JSON:

• JSON データの列分割を指定するには、[詳細]>[入れ子のレベル] を 1 から 100 までで選択します。

• Advanced>Ignore データ形式エラーを選択した場合、データは JSON 形式で取り込まれます。 このチェック ボックスをオフのままにすると、データは multijson 形式で取り込まれます。

  

まとめ

[データ準備] ウィンドウでは、データ インジェストが正常に終了した場合、3 つのステップすべてに緑色のチェックマークが表示されます。 各ステップで使用されたコマンドを表示したり、取り込まれたデータのクエリ、視覚化、削除を行うカードを選択したりできます。

ポータル - Azure Data Explorer ページ

前提条件

• Azure サブスクリプション。 無料の Azure アカウントを作成します。
• Azure Data Explorer クラスターとデータベース。 クラスターとデータベースを作成します。
• 宛先テーブル。 テーブルを作成するか、既存のテーブルを使用する。
• テーブルのインジェスト マッピング。
• インジェスト対象のデータが存在するイベント ハブ。

データを取得する

1. Azure portal でクラスターに移動し、 Databases を選択します。 次に、ターゲット テーブルを含むデータベースを選択します。

   

2. 左側のメニューから、 Data インジェストを選択します。 次に、上部のバーで[データ接続の追加] 選択。

3. フォームに次の情報を入力し、 [作成] を選択します。

   設定	推奨値	フィールドの説明
   データ接続名	test-hub-connection	Azure データ エクスプローラーで作成する接続の名前。
   サブスクリプション		イベントハブ リソースが配置されているサブスクリプション ID。
   イベント ハブの名前空間	一意の名前空間名	以前に選択した、名前空間を識別する名前。
   イベント ハブ	test-hub	作成したイベント ハブ。
   コンシューマー グループ	test-group	作成したイベント ハブに定義されているコンシューマー グループ。
   イベント システム プロパティ	関連するプロパティを選択する	イベント ハブのシステム プロパティ。 イベント メッセージごとに複数のレコードがある場合は、システム プロパティが最初のレコードに追加されます。 システム プロパティを追加する場合は、テーブル スキーマとマッピングを作成または更新して、選択したプロパティを含めます。
   圧縮	なし	イベント ハブ メッセージ ペイロードの圧縮の種類。 サポートされている圧縮の種類: None、gzip。
   マネージド ID (推奨)	システム割り当て	イベント ハブに対する読み取りアクセスのために Data Explorer クラスターによって使用されるマネージド ID。 イベント ハブへのアクセスを制御するために、マネージド ID を使用することをお勧めします。 注: データ接続が作成されると、次のようになります。 * "システム割り当て" ID が存在しない場合は自動的に作成されます * マネージド ID は、"Azure Event Hubs データ受信者" ロールが自動的に割り当てられ、Data Explorer クラスターに追加されます。 ロールが割り当てられ、ID がクラスターに追加されたことを確認することをお勧めします。

   Note

   マネージド ID を使用していない既存のデータ接続がある場合は、マネージド ID を使用するように更新することをお勧めします。

ポータル - Azure Event Hubs ページ

前提条件

• Azure サブスクリプション。 無料の Azure アカウントを作成します。
• Azure Data Explorer クラスターとデータベース。 クラスターとデータベースを作成します。
• 宛先テーブル。 テーブルを作成するか、既存のテーブルを使用する。
• テーブルのインジェスト マッピング。
• インジェスト対象のデータが存在するイベント ハブ。

データを取得する

1. Azure portal で、Event Hubs インスタンスを参照します。

2. [ Features サイド メニューの [Kusto でデータ Analyze] を選択。

3. Start を選択して、Azure Data Explorer へのインジェスト ウィザードを開きます。

   

   Note

   • 少なくとも Database User アクセス許可が必要です。
   • パブリック アクセスなしでクラスターとストレージ アカウント間のアクセスを有効にするには、「 マネージド プライベート エンドポイントの作成を参照してください。
   • クラスターとイベント ハブは、同じテナントに関連付ける必要があります。 そうでない場合は、C# や Python などの SDK オプションのいずれかを使用します。

4. [ Ingest data ] サイド ウィンドウが開き、[ Destination ] タブが選択されています。 ドロップダウンから Cluster フィールドと Database フィールドを選択します。 実行しているクラスターを選択してください。 それ以外の場合は、[データベース] を選択してインジェスト プロセスを続行することはできません。

5. [ テーブル]で[新しいテーブル 選択し 新しいテーブルの名前を入力します。 または、 Existing テーブルを使用します。

6. [Next: Source](次へ: ソース) を選択します。

7. Source の種類では、Event Hub の種類と詳細は、開始元の Event Hubs インスタンスに基づいて自動的に設定されます。

8. [ Data 接続で、次のフィールドに入力し、 Next: Schema を選択します。

   設定	推奨値	フィールドの説明
   サブスクリプション		イベントハブ リソースが配置されているサブスクリプション ID。
   イベント ハブの名前空間		名前空間を識別する名前。
   イベント ハブ		使用するイベント ハブ。
   データ接続名	TestDataConnection	データ接続を識別する名前。
   コンシューマー グループ		イベント ハブに定義されているコンシューマー グループ。
   Compression		イベント ハブ メッセージ ペイロードの圧縮の種類。
   イベント システム プロパティ	関連するプロパティを選択する	イベント ハブのシステム プロパティ。 イベント メッセージごとに複数のレコードがある場合、システム プロパティは最初のものに追加されます。 システム プロパティを追加する場合は、テーブル スキーマとマッピングを作成または更新して、選択したプロパティを含めます。
   イベント取得の開始日	協定世界時 (UTC)	データ接続は、 Event 取得の開始日後に作成された既存の Event Hubs イベントを取得します。 取得できるのは、Event Hubs のリテンション期間によって保持されるイベントのみです。 Event 取得の開始日が指定されていない場合、既定の時刻はデータ接続が作成される時刻です。

9. クラスターに対して streaming が有効になっている場合は、 ストリーミング インジェストを選択できます。 クラスターでストリーミングが有効になっていない場合は、 Data バッチ処理の待機時間を設定します。 Event Hubs の場合、推奨される バッチ処理時間 は 30 秒です。

   

10. Data 形式を選択します。 CSV 形式のデータの場合 ファイルの見出し行を無視するには 最初のレコードを配置します。 JSON 形式のデータの場合は、 [データ形式のエラー を選択して JSON 形式でデータを取り込むか、選択を解除して multijson 形式でデータを取り込みます。 Nested レベルを選択して、テーブル列のデータ除算を決定します。

11. プレビュー ウィンドウに表示されるデータが完全でない場合は、必要なすべてのデータ フィールドを含むテーブルを作成するために、さらにデータが必要になることがあります。 次のコマンドを使用して、イベント ハブから新しいデータをフェッチします。

    • Discard and fetch new data (破棄して新しいデータをフェッチする) : 表示されたデータを破棄し、新しいイベントを検索します。
    • Fetch more data (さらにデータをフェッチする) :既に見つかったイベントに加えて、さらにイベントを検索します。

    注意

    データのプレビューを表示するには、イベント ハブがイベントを送信している必要があります。

12. 概要 を選択します。

13. 確立された Event Hub からの継続的なインジェスト ウィンドウでは、確立が正常に完了すると、すべての手順が緑色のチェック マークでマークされます。

ARM テンプレート

前提条件

• Azure サブスクリプション。 無料の Azure アカウントを作成します。
• Azure Data Explorer クラスターとデータベース。 クラスターとデータベースを作成します。
• 宛先テーブル。 テーブルを作成するか、既存のテーブルを使用する。
• テーブルのインジェスト マッピング。
• インジェスト対象のデータが存在するイベント ハブ。

ARM テンプレート

次の例は、Event Hubs データ接続を追加する Azure Resource Manager テンプレートを示しています。 フォームを使用して、Azure portal でテンプレートの編集とデプロイを実行できます。

{
    "$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "parameters": {
        "namespaces_eventhubns_name": {
            "type": "string",
            "defaultValue": "eventhubns",
            "metadata": {
                "description": "Specifies the Event Hubs Namespace name."
            }
        },
        "EventHubs_eventhubdemo_name": {
            "type": "string",
            "defaultValue": "eventhubdemo",
            "metadata": {
                "description": "Specifies the event hub name."
            }
        },
        "consumergroup_default_name": {
            "type": "string",
            "defaultValue": "$Default",
            "metadata": {
                "description": "Specifies the consumer group of the event hub."
            }
        },
        "Clusters_kustocluster_name": {
            "type": "string",
            "defaultValue": "kustocluster",
            "metadata": {
                "description": "Specifies the name of the cluster"
            }
        },
        "databases_kustodb_name": {
            "type": "string",
            "defaultValue": "kustodb",
            "metadata": {
                "description": "Specifies the name of the database"
            }
        },
        "tables_kustotable_name": {
            "type": "string",
            "defaultValue": "kustotable",
            "metadata": {
                "description": "Specifies the name of the table"
            }
        },
        "mapping_kustomapping_name": {
            "type": "string",
            "defaultValue": "kustomapping",
            "metadata": {
                "description": "Specifies the name of the mapping rule"
            }
        },
        "dataformat_type": {
            "type": "string",
            "defaultValue": "csv",
            "metadata": {
                "description": "Specifies the data format"
            }
        },
        "databaseRouting_type": {
            "type": "string",
            "defaultValue": "Single",
            "metadata": {
                "description": "The database routing for the connection. If you set the value to **Single**, the data connection will be routed to a single database in the cluster as specified in the *databaseName* setting. If you set the value to **Multi**, you can override the default target database using the *Database* EventData property."
            }
        },
        "dataconnections_kustodc_name": {
            "type": "string",
            "defaultValue": "kustodc",
            "metadata": {
                "description": "Name of the data connection to create"
            }
        },
        "subscriptionId": {
            "type": "string",
            "defaultValue": "[subscription().subscriptionId]",
            "metadata": {
                "description": "Specifies the subscriptionId of the event hub"
            }
        },
        "resourceGroup": {
            "type": "string",
            "defaultValue": "[resourceGroup().name]",
            "metadata": {
                "description": "Specifies the resourceGroup of the event hub"
            }
        },
        "location": {
            "type": "string",
            "defaultValue": "[resourceGroup().location]",
            "metadata": {
                "description": "Location for all resources."
            }
        }
    },
    "variables": {
    },
    "resources": [{
            "type": "Microsoft.Kusto/Clusters/Databases/DataConnections",
            "apiVersion": "2022-02-01",
            "name": "[concat(parameters('Clusters_kustocluster_name'), '/', parameters('databases_kustodb_name'), '/', parameters('dataconnections_kustodc_name'))]",
            "location": "[parameters('location')]",
            "kind": "EventHub",
            "properties": {
                "managedIdentityResourceId": "[resourceId('Microsoft.Kusto/clusters', parameters('clusters_kustocluster_name'))]",
                "eventHubResourceId": "[resourceId(parameters('subscriptionId'), parameters('resourceGroup'), 'Microsoft.EventHub/namespaces/eventhubs', parameters('namespaces_eventhubns_name'), parameters('EventHubs_eventhubdemo_name'))]",
                "consumerGroup": "[parameters('consumergroup_default_name')]",
                "tableName": "[parameters('tables_kustotable_name')]",
                "mappingRuleName": "[parameters('mapping_kustomapping_name')]",
                "dataFormat": "[parameters('dataformat_type')]",
                "databaseRouting": "[parameters('databaseRouting_type')]"
            }
        }
    ]
}

イベント ハブ データ接続を削除する

データの取得

ポータル タブで説明されているように、Azure portal を使用してデータ接続を削除します。

ポータル

Azure portal からイベント ハブ接続を削除するには、次の操作を行います。

1. クラスターに移動します。 左側のメニューで [データベース] を選択します。 対象となるテーブルを含むデータベースを選択します。
2. 左側のメニューで [データ接続] を選択します。 次に、関連するイベント ハブ データ接続の横にあるチェック ボックスをオンにします。
3. 上部のメニュー バーで [削除] を選択します。

関連するコンテンツ

• イベント ハブ サンプル メッセージ アプリを使用して接続を確認する
• Web UI でデータに対してクエリを実行する