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

[アーティクル]
03/05/2024

この記事では、Event Grid データ接続を使用して、ストレージアカウントから Azure Data Explorer に BLOB を取り込む方法を説明します。 Azure Event Grid サブスクリプションを設定する Event Grid データ接続を作成します。 Event Grid サブスクリプションは、Azure Event Hubs 経由でストレージアカウントから Azure Data Explorer にイベントをルーティングします。

注意

インジェストでは、最大で 6 GB のファイルサイズがサポートされます。 100 MB から 1 GB の間のファイルを取り込むことをお勧めします。

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

Event Grid から Azure Data Explorer への取り込みに関する一般的な情報については、Event Grid への接続に関する記事を参照してください。

注意

Event Grid 接続で最高のパフォーマンスを実現するには、BLOB メタデータを rawSizeBytes 使用してインジェストプロパティを設定します。詳細については、「インジェストプロパティ」を参照してください。

前提条件

Azure サブスクリプション。無料の Azure アカウントを作成します。
Azure Data Explorer クラスターとデータベース。クラスターとデータベースを作成します。
変換先テーブル。テーブルを作成するか、既存のテーブルを使用します。
テーブルのインジェストマッピング。
ストレージアカウント。 Event Grid 通知サブスクリプションは、または Data Lake Storage Gen2の BlobStorageStorageV2Azure Storage アカウントで設定できます。
Event Grid リソースプロバイダーを登録します。

Event Grid データ接続を作成する

このセクションでは、Event Grid と Azure Data Explorer テーブル間の接続を確立します。

Azure portalで Azure Data Explorer クラスターを参照します。
[データ] で、[データベース] [テスト][>データベース] の順に選択します。
[ 設定] で [ データ接続 ] を選択し、[ データ接続>の追加] [Event Grid (Blob Storage)] の順に選択します。

Event Grid データ接続フォームに次の情報を入力します。

設定	推奨値	フィールドの説明
データ接続名	test-grid-connection	Azure Data Explorer で作成する接続の名前。データ接続名には、英数字、ダッシュ、ドット文字のみを含め、最大 40 文字の長さにできます。
ストレージアカウントのサブスクリプション	サブスクリプション ID	ストレージアカウントが存在するサブスクリプション ID。
イベントの種類	"作成された BLOB" または "名前変更された BLOB"	インジェストをトリガーするイベントの種類。 "名前変更された BLOB" は、ADLSv2 ストレージに対してのみサポートされています。 BLOB の名前を変更するには、Azure portal内の BLOB に移動し、BLOB を右クリックして [名前の変更] を選択します。サポートされる種類は、Microsoft.Storage.BlobCreated または Microsoft.Storage.BlobRenamed。
ストレージアカウント	gridteststorage1	作成済みのストレージアカウントの名前。
リソースの作成	自動	リソースの自動作成を有効にすると、Azure Data Explorerによって Event Grid サブスクリプション、Event Hubs 名前空間、および Event Hubs が自動的に作成されます。それ以外の場合は、これらのリソースを手動で作成して、データ接続を確実に作成する必要があります。「Event Grid インジェスト用のリソースを手動で作成する」を参照してください

必要に応じて、特定の Event Grid の件名を追跡できます。次のように、通知用のフィルターを設定します。
- [Prefix](プレフィックス) フィールドはサブジェクトの "リテラル" プレフィックスです。適用されるパターンは で始まるので、複数のコンテナー、フォルダー、または BLOB にまたがることができます。ワイルドカードは使用できません。
  - BLOB コンテナーに対してフィルターを定義するには、フィールドをのように設定する "必要があります"。
  - BLOB プレフィックス (または Azure Data Lake Gen2 のフォルダー) に対してフィルターを定義するには、フィールドをのように設定する "必要があります"。
- [Suffix](サフィックス) フィールドは、BLOB の "リテラル" サフィックスです。ワイルドカードは使用できません。
- [Case-Sensitive](大文字と小文字の区別) フィールドは、プレフィックスとサフィックスのフィルターで大文字と小文字が区別されるかどうかを示します
イベントのフィルター処理の詳細については、「 Blob Storage イベント」を参照してください。

必要に応じて、次の情報に従って データルーティング設定 を指定できます。 データのルーティング設定はすべて指定する必要はありません。部分的な設定も受け入れられます。

設定	推奨値	フィールドの説明
他のデータベースへのデータのルーティングを許可する (マルチデータベースデータ接続)	Don't allow	データ接続に関連付けられている既定のターゲットデータベースをオーバーライドする場合は、このオプションをオンに切り替えます。データベースのルーティングの詳細については、「イベントルーティング」を参照してください。
テーブル名	TestTable	TestDatabase に作成したテーブル。
データ形式	JSON	サポートされている形式は、APACHEAVRO、Avro、CSV、JSON、ORC、PARQUET、PSV、RAW、SCSV、SOHSV、TSV、TSVE、TXT、W3CLOG です。サポートされている圧縮オプションは、Zip と Gzip です。
マッピング名	TestTable_mapping	TestDatabase に作成したマッピング。これにより、受信データを TestTable の列名とデータ型にマッピングします。指定しない場合、テーブルのスキーマから派生した ID データマッピングが自動生成されます。
書式エラーを無視する	無視	JSON データ形式の形式エラーを無視する場合は、このオプションをオンに切り替えます。

注意

テーブル名とマッピング名では、大文字と小文字が区別されます。

必要に応じて、[ 詳細設定] で、データ接続で使用される マネージド ID の種類 を指定できます。既定では、[ システム割り当て] が選択されています。

[ ユーザー割り当て] を選択した場合は、マネージド ID を手動で割り当てる必要があります。クラスターにまだ割り当てられていないユーザーを選択すると、自動割り当てされます。詳細については、「Azure Data Explorer クラスターのマネージド ID を構成する」を参照してください。

[なし] を選択した場合、ストレージアカウントとイベントハブは接続文字列を介して認証されます。この方法は推奨されません。

[作成]

Azure portalでストレージアカウントを参照します。
イベントを選択します。
[作業の開始] タブで、[Azure Data Explorer] タブを選択します。
[ 継続的なデータインジェストの作成 ] タイルで、[作成] を選択 します。

イベントをルーティングする宛先を選択する

データを取り込むクラスターを選択します。
ターゲットデータベースを選択します。
[選択] を選択します。

データ接続を作成する

フォームに次の情報を入力します。

設定	推奨値	フィールドの説明
データ接続名	test-grid-connection	Azure Data Explorer で作成する接続の名前。
ストレージアカウントのサブスクリプション	サブスクリプション ID	ストレージアカウントが存在するサブスクリプション ID。
イベントの種類	"作成された BLOB" または "名前変更された BLOB"	インジェストをトリガーするイベントの種類。 "名前変更された BLOB" は、ADLSv2 ストレージに対してのみサポートされています。 BLOB の名前を変更するには、Azure portal内の BLOB に移動し、BLOB を右クリックして [名前の変更] を選択します。サポートされる種類は、Microsoft.Storage.BlobCreated または Microsoft.Storage.BlobRenamed。
ストレージアカウント	gridteststorage1	このウィザードにアクセスしたストレージアカウントが自動的に設定されます。
リソースの作成	自動	リソースの自動作成を有効にすると、Azure Data Explorerによって Event Grid サブスクリプション、Event Hubs 名前空間、および Event Hubs が自動的に作成されます。それ以外の場合は、これらのリソースを手動で作成して、データ接続を確実に作成する必要があります。「Event Grid インジェスト用のリソースを手動で作成する」を参照してください

必要に応じて、特定の Event Grid の件名を追跡できます。次のように、通知用のフィルターを設定します。
- [Prefix](プレフィックス) フィールドはサブジェクトの "リテラル" プレフィックスです。適用されるパターンは で始まるので、複数のコンテナー、フォルダー、または BLOB にまたがることができます。ワイルドカードは使用できません。
  - BLOB コンテナーに対してフィルターを定義するには、フィールドをのように設定する "必要があります"。
  - BLOB プレフィックス (または Azure Data Lake Gen2 のフォルダー) に対してフィルターを定義するには、フィールドをのように設定する "必要があります"。
- [Suffix](サフィックス) フィールドは、BLOB の "リテラル" サフィックスです。ワイルドカードは使用できません。
- [Case-Sensitive](大文字と小文字の区別) フィールドは、プレフィックスとサフィックスのフィルターで大文字と小文字が区別されるかどうかを示します
イベントのフィルター処理の詳細については、「 Blob Storage イベント」を参照してください。

設定	推奨値	フィールドの説明
他のデータベースへのデータのルーティングを許可する (マルチデータベースデータ接続)	Don't allow	データ接続に関連付けられている既定のターゲットデータベースをオーバーライドする場合は、このオプションをオンに切り替えます。データベースのルーティングの詳細については、「イベントルーティング」を参照してください。
テーブル名	TestTable	TestDatabase に作成したテーブル。
データ形式	JSON	サポートされている形式は、APACHEAVRO、Avro、CSV、JSON、ORC、PARQUET、PSV、RAW、SCSV、SOHSV、TSV、TSVE、TXT、W3CLOG です。サポートされている圧縮オプションは、Zip と Gzip です。
マッピング名	TestTable_mapping	TestDatabase に作成したマッピング。これにより、受信データを TestTable の列名とデータ型にマッピングします。指定しない場合、テーブルのスキーマから派生した ID データマッピングが自動生成されます。
書式エラーを無視する	無視	JSON データ形式の形式エラーを無視する場合は、このオプションをオンに切り替えます。

注意

テーブル名とマッピング名では、大文字と小文字が区別されます。

必要に応じて、[ 詳細設定] で、データ接続で使用される マネージド ID の種類 を指定できます。既定では、[ システム割り当て] が選択されています。

[ ユーザー割り当て] を選択した場合は、マネージド ID を手動で割り当てる必要があります。クラスターにまだ割り当てられていないユーザーを選択すると、自動割り当てされます。詳細については、「Azure Data Explorer クラスターのマネージド ID を構成する」を参照してください。

[なし] を選択した場合、ストレージアカウントとイベントハブは接続文字列を介して認証されます。この方法は推奨されません。

［作成］ を選択します

次の例は、Event Grid データ接続を追加するための 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 hub 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."
            }
        },
        "StorageAccounts_storagedemo_name": {
            "type": "string",
            "defaultValue": "storagedemo",
            "metadata": {
                "description": "Specifies the storage account name"
            }
        },
        "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 resources"
            }
        },
        "resourceGroup": {
            "type": "string",
            "defaultValue": "[resourceGroup().name]",
            "metadata": {
                "description": "Specifies the resourceGroup of the resources"
            }
        },
        "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": "EventGrid",
            "properties": {
                "managedIdentityResourceId": "[resourceId('Microsoft.Kusto/clusters', parameters('clusters_kustocluster_name'))]",
                "storageAccountResourceId": "[resourceId(parameters('subscriptionId'), parameters('resourceGroup'), 'Microsoft.Storage/storageAccounts', parameters('StorageAccounts_storagedemo_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')]"
            }
        }
    ]
}

Event Grid データ接続を使用する

このセクションでは、BLOB の作成または BLOB の名前変更に従って、Azure Blob Storageまたは Azure Data Lake Gen 2 からクラスターへのインジェストをトリガーする方法について説明します。

BLOB のアップロードに使用されるストレージ SDK の種類に基づいて、関連するタブを選択します。

Azure Blob Storage
Azure Data Lake

次のコードサンプルでは、Azure Blob Storage SDK を使用してファイルをAzure Blob Storageにアップロードします。アップロードによって Event Grid データ接続がトリガーされ、Azure Data Explorerにデータが取り込まれます。

var azureStorageAccountConnectionString = <storage_account_connection_string>;
var containerName = <container_name>;
var blobName = <blob_name>;
var localFileName = <file_to_upload>;
var uncompressedSizeInBytes = <uncompressed_size_in_bytes>;
var mapping = <mapping_reference>;
// Create a new container if it not already exists.
var azureStorageAccount = new BlobServiceClient(azureStorageAccountConnectionString);
var container = azureStorageAccount.GetBlobContainerClient(containerName);
container.CreateIfNotExists();
// Define blob metadata and uploading options.
IDictionary<String, String> metadata = new Dictionary<string, string>();
metadata.Add("rawSizeBytes", uncompressedSizeInBytes);
metadata.Add("kustoIngestionMappingReference", mapping);
var uploadOptions = new BlobUploadOptions
{
    Metadata = metadata,
};
// Upload the file.
var blob = container.GetBlobClient(blobName);
blob.Upload(localFileName, uploadOptions);

注意

Azure Data Explorer では、BLOB 投稿の取り込みは削除されません。 BLOB の削除を管理する Azure Blob Storage のライフサイクルを使用して、BLOB を 3 から 5 日間保持します。

次のコードサンプルでは、Azure Data Lake SDK を使用してファイルをData Lake Storage Gen2にアップロードします。アップロードによって Event Grid データ接続がトリガーされ、Azure Data Explorerにデータが取り込まれます。

var accountName = <storage_account_name>;
var accountKey = <storage_account_key>;
var fileSystemName = <file_system_name>;
var fileName = <file_name>;
var localFileName = <file_to_upload>;
var uncompressedSizeInBytes = <uncompressed_size_in_bytes>;
var mapping = <mapping_reference>;
var sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey);
var dfsUri = "https://" + accountName + ".dfs.core.windows.net";
var dataLakeServiceClient = new DataLakeServiceClient(new Uri(dfsUri), sharedKeyCredential);
// Create the filesystem.
var dataLakeFileSystemClient = dataLakeServiceClient.CreateFileSystem(fileSystemName).Value;
// Define file metadata and uploading options.
IDictionary<String, String> metadata = new Dictionary<string, string>();
metadata.Add("rawSizeBytes", uncompressedSizeInBytes);
metadata.Add("kustoIngestionMappingReference", mapping);
var uploadOptions = new DataLakeFileUploadOptions
{
    Metadata = metadata,
    Close = true // Note: The close option triggers the event being processed by the data connection.
};
// Upload the file.
var dataLakeFileClient = dataLakeFileSystemClient.GetFileClient(fileName);
dataLakeFileClient.Upload(localFileName, uploadOptions);

注意

Azure Data Lake SDK を使用してファイルをアップロードする場合、最初のファイル作成イベントのサイズは 0 です。これは、データインジェスト中に Azure Data Explorerによって無視されます。適切なインジェストを確実に行うために、パラメーターを Close に true設定します。このパラメーターを指定すると、upload メソッドは FlushAndClose イベントをトリガーし、最終的な更新が行われ、ファイルストリームが閉じられたことを示します。
Event Grid からのトラフィックを減らし、Azure Data Explorerへのイベントの取り込みを最適化するには、data.api キーをフィルター処理して CreateFile イベントを除外することをお勧めします。これにより、サイズ 0 のファイル作成イベントが除外され、空のファイルのインジェストエラーが回避されます。フラッシュについて詳しくは、Azure Data Lake の Flush メソッドに関するページを参照してください。

BLOB を名前変更する

ADLSv2 では、ディレクトリの名前を変更できます。ただし、ディレクトリの名前を変更しても、BLOB の名前が変更されたイベントがトリガーされたり、ディレクトリ内に含まれる BLOB のインジェストが開始されたりすることはありません。ディレクトリの名前を変更した後に BLOB の取り込みを確実に行う場合は、ディレクトリ内の個々の BLOB の名前を直接変更する必要があります。

次のコードサンプルは、ADLSv2 ストレージアカウント内の BLOB の名前を変更する方法を示しています。

var accountName = <storage_account_name>;
var accountKey = <storage_account_key>;
var fileSystemName = <file_system_name>;
var sourceFilePath = <source_file_path>;
var destinationFilePath = <destination_file_path>;
var sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey);
var dfsUri = "https://" + accountName + ".dfs.core.windows.net";
var dataLakeServiceClient = new DataLakeServiceClient(new Uri(dfsUri), sharedKeyCredential);
// Get a client to the the filesystem

var dataLakeFileSystemClient = dataLakeServiceClient.GetFileSystemClient(fileSystemName);
// Rename a file in the file system
var dataLakeFileClient = dataLakeFileSystemClient.GetFileClient(sourceFilePath);
dataLakeFileClient.Rename(destinationFilePath);

注意

データ接続を作成するとき、または Event Grid リソースを手動で作成するときに、特定のサブジェクトを追跡するフィルターを定義した場合、これらのフィルターは対象のファイルパスに適用されます。

注意

操作に続くインジェストの CopyBlob トリガーは、階層型名前空間機能が有効になっているストレージアカウントではサポートされていません。

Event Grid データ接続を削除する

Azure portalから Event Grid 接続を削除するには、次の手順を実行します。

クラスターに移動します。左側のメニューから [データベース] を選択 します。次に、ターゲットテーブルを含むデータベースを選択します。
左側のメニューで、[ データ接続] を選択します。次に、関連する Event Grid データ接続の横にあるチェックボックスをオンにします。
上部のメニューバーで、[削除] を選択 します。

Azure Stream Analytics を使用してイベントハブからデータを処理する

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

前提条件

Event Grid データ接続を作成する

イベントをルーティングする宛先を選択する

データ接続を作成する

Event Grid データ接続を使用する

BLOB を名前変更する

Event Grid データ接続を削除する

フィードバック

フィードバック

その他のリソース

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

前提条件

Event Grid データ接続を作成する

Event Grid データ接続を使用する

Event Grid データ接続を削除する

関連コンテンツ

フィードバック

フィードバック

その他のリソース