Azure Stream Analytics から Azure Data Explorer にデータを取り込む

重要

このコネクタは、Microsoft Fabric の リアルタイム分析 で使用できます。 次の例外を含め、この記事の手順を使用します。

• 必要に応じて、「KQL データベースを作成する」の手順 に従ってデータベースを作成します。
• 必要に応じて、「空のテーブルを作成する」の手順に従って テーブルを作成します。
• 「コピー URI」の手順を使用して、クエリまたはインジェスト URI を取得します。
• KQL クエリセットでクエリを実行します。

Azure Data Explorer では、Azure Stream Analytics からのデータ インジェストがサポートされています。 Azure Stream Analytics は、複数のソースから大量の高速ストリーミング データを同時に処理するように設計された、リアルタイム分析および複雑なイベント処理エンジンです。

Azure Stream Analytics "ジョブ" は、入力ソース、変換クエリ、出力接続で構成されます。 Azure portal、Azure Resource Manager (ARM) テンプレート、Azure PowerShell、.NET API、REST API、Visual Studio、Stream Analytics no コード エディターを使用して、Stream Analytics ジョブを作成、編集、テストできます。

この記事では、Stream Analytics ジョブを使用してイベント ハブからデータを収集し、Azure portalまたは ARM テンプレートを使用して Azure Data Explorer クラスターに送信する方法について説明します。

前提条件

• クラスターとデータベースおよびテーブルを作成します。

• Azure Stream Analytics チュートリアルの次のセクションを使用して、イベント ハブを作成します。

  • イベント ハブの作成
  • イベント ハブへのアクセスを許可し、接続文字列を取得する

  ヒント

  テストの目的で、Microsoft ダウンロード センターから通話イベント ジェネレーター アプリをダウンロードするか、GitHub からソース コードを入手することをお勧めします。 Azure Stream Analytics ジョブを設定するときに、イベント ハブからデータをプルし、Azure Data Explorer 出力コネクタに渡すように構成します。

Azure Data Explorer 出力接続を作成する

次の手順を使用して、Azure portal または ARM テンプレートを使用して Stream Analytics ジョブの Azure Data Explorer 出力を作成します。 接続は、指定した Azure Data Explorer テーブルにデータを送信するために Stream Analytics ジョブによって使用されます。 作成され、ジョブが実行されると、ジョブに流れ込むデータは、指定されたターゲット テーブルに取り込まれます。

重要

• Azure Data Explorer 出力コネクタでは、マネージド ID 認証のみがサポートされます。 コネクタの作成の一環として、データベース モニターとデータベース インジェスターのアクセス許可が Azure Stream Analytics ジョブのマネージド ID に付与されます。
• Azure Data Explorer 出力コネクタを設定するときに、ターゲット クラスター、データベース、およびテーブル名を指定します。 インジェストを成功させるには、Azure Stream Analytics クエリで定義されているすべての列が、Azure Data Explorer テーブルの列名と型と一致している必要があります。 列名は大文字と小文字が区別され、任意の順序で指定できます。 Azure Stream Analytics クエリに、Azure Data Explorer テーブルの列にマップされない列がある場合は、エラーが発生します。

注意

• Azure Stream Analytics のすべての入力がサポートされています。 コネクタは入力を CSV 形式に変換し、指定した Azure Data Explorer テーブルにデータをインポートします。
• Azure Data Explorer には、インジェスト プロセスを最適化することを目的とした、データ インジェストの集計 (バッチ処理) ポリシーがあります。 既定では、このポリシーは 5 分、1000 項目、または 1GB のデータに構成されているため、待ち時間が生じることがあります。 集計オプションの構成については、「バッチ処理のポリシー」を参照してください。

Azure Portal

開始する前に、既存の Stream Analytics ジョブがあることを確認するか、新しいジョブを作成してから、次の手順に従って Azure Data Explorer 接続を作成します。

1. Azure portal にサインインします。

2. Azure portal から [すべてのリソース] を開き、使用する Stream Analytics ジョブを選択します。

3. [ジョブ トポロジ] で、[出力] を選択します。

4. [Azure Data Explorerの追加] を>選択します。

   

5. 次の情報を使用して出力フォームに入力し、[保存] を選択します。

   注意

   クラスターとデータベースを指定するために、次のオプションを使用できます。

   • サブスクリプション: [Select Azure Data Explorer from your subscriptions] (サブスクリプションから Azure Data Explorer を選択する) を選択し、サブスクリプションを選択してから、クラスターとデータベースを選択します。
   • 手動: [Provide Azure Data Explorer settings manually] (Azure Data Explorer の設定を手動で指定する) を選択し、クラスターの URI とデータベースを指定します。

   プロパティ名	説明
   出力エイリアス	クエリの出力をこのデータベースに出力するためにクエリで使用されるわかりやすい名前です。
   サブスクリプション	クラスターが存在する Azure サブスクリプションを選択します。
   クラスター	クラスターを識別する一意の名前。 指定したクラスター名に、ドメイン名 [region].kusto.windows.net が付加されます。 名前に使用できるのは、小文字と数字だけです。 4 文字以上 22 文字以内にする必要があります。
   クラスター URI	クラスターのデータ インジェスト URI。 Azure Data Explorer または Azure Synapse データ エクスプローラーのデータ インジェスト エンドポイントの URI を指定できます。
   データベース	ご自分の出力を送信するデータベースの名前です。 データベース名はクラスター内で一意である必要があります。
   認証	クラスターが他のMicrosoft Entra保護されたリソースに簡単にアクセスできるようにするMicrosoft Entraマネージド ID。 ID は Azure プラットフォームによって管理され、シークレットをプロビジョニングまたはローテーションする必要はありません。 マネージド ID 構成により、クラスターにカスタマー マネージド キーを使用できます。
   テーブル	出力を送信するテーブルの名前。 Azure Stream Analytics 出力の列名とデータ型は、Azure Data Explorer テーブルのスキーマと一致している必要があります。

   

Azure Resource Manager テンプレート

次の例は、Azure Data Explorer 出力コネクタを追加する Azure Resource Manager テンプレートを示しています。 Azure portal でテンプレートを編集してデプロイし、コネクタを作成できます。

{
    "$schema": "http://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "parameters": {
        "streamingJobName": {
            "type": "string",
            "defaultValue": "streamingjob"
        },
        "eventHubNamespace": {
            "type": "string"
        },
        "eventHubKey": {
            "type": "securestring"
        },
        "eventHubName": {
            "type": "string"
        },
        "eventHubConsumerGroupName": {
            "type": "string"
        },
        "streamingUnits": {
            "type": "int",
            "defaultValue": 3
        },
        "ADXClusterUri": {
            "type": "string"
        },
        "ADXDatabaseName": {
            "type": "string"
        },
        "ADXTableName": {
            "type": "string"
        }
    },
    "resources": [
        {
            "apiVersion": "2017-04-01-preview",
            "name": "[parameters('streamingJobName')]",
            "location": "[resourceGroup().location]",
            "type": "Microsoft.StreamAnalytics/StreamingJobs",
            "identity": {
                "type": "systemAssigned"
            },
            "properties": {
                "sku": {
                    "name": "standard"
                },
                "eventsOutOfOrderPolicy": "drop",
                "eventsOutOfOrderMaxDelayInSeconds": 10,
                "compatibilityLevel": "1.2",
                "outputStartMode": "JobStartTime",
                "inputs": [
                    {
                        "name": "inputEventHub",
                        "properties": {
                            "type": "stream",
                            "serialization": {
                                "type": "JSON",
                                "properties": {
                                    "encoding": "UTF8"
                                }
                            },
                            "datasource": {
                                "type": "Microsoft.ServiceBus/EventHub",
                                "properties": {
                                    "serviceBusNamespace": "[parameters('eventHubNamespace')]",
                                    "sharedAccessPolicyName": "RootManageSharedAccessKey",
                                    "sharedAccessPolicyKey": "[parameters('eventHubKey')]",
                                    "eventHubName": "[parameters('eventHubName')]",
                                    "consumerGroupName": "[parameters('eventHubConsumerGroupName')]"
                                }
                            }
                        }
                    }
                ],
                "transformation": {
                    "name": "ProcessSampleData",
                    "properties": {
                        "streamingUnits": "[int(parameters('streamingUnits'))]",
                        "query": "select * from inputEventHub"
                    }
                },
                "outputs": [
                    {
                        "name": "output",
                        "properties": {
                            "serialization": {
                                "type": "CSV",
                                "properties": {
                                    "encoding": "UTF8"
                                }
                            },
                            "datasource": {
                                "type": "Microsoft.Kusto/clusters/databases",
                                "properties": {
                                    "cluster": "[parameters('ADXClusterUri')]",
                                    "table": "[parameters('ADXTableName')]",
                                    "database": "[parameters('ADXDatabaseName')]",
                                    "authenticationMode": "Msi"
                                }
                            }
                        }
                    }
                ]
            }
        }
    ]
}

注意

ARM テンプレートを使用して Azure Data Explorer 出力接続を作成する場合、Azure Data Explorer データベースに対する Streaming Analytics ジョブのマネージド ID アクセス許可を手動で付与する必要があります。

関連コンテンツ

• Azure Data Explorer でデータのクエリを実行する
• Azure Stream Analytics ジョブから Azure Functions を実行する