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

1. Azure portal で、Azure Data Explorer クラスターに移動します。

2. [データ] で [データベース]>[TestDatabase] を選択します。

   

3. [設定] で [データ接続] を選択し、[データ接続の追加]>[Event Grid (Blob Storage)] を選択します。

   

4. 次の情報を使用して、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 インジェスト用のリソースを手動で作成する」を参照してください。

   1. 必要に応じて、特定の Event Grid の件名を追跡できます。 次のように、通知用のフィルターを設定します。

      • [Prefix](プレフィックス) フィールドはサブジェクトの "リテラル" プレフィックスです。 適用されるパターンは [次の文字列で始まる] で、複数のコンテナー、フォルダー、BLOB を対象にできます。 ワイルドカードは使用できません。
        • BLOB コンテナーに対してフィルターを定義するには、フィールドを /blobServices/default/containers/[container prefix] のように設定する "必要があります"。
        • BLOB プレフィックス (または Azure Data Lake Gen2 のフォルダー) に対してフィルターを定義するには、フィールドを /blobServices/default/containers/[container name]/blobs/[folder/blob prefix] のように設定する "必要があります"。
      • [Suffix](サフィックス) フィールドは、BLOB の "リテラル" サフィックスです。 ワイルドカードは使用できません。
      • [Case-Sensitive](大文字と小文字の区別) フィールドは、プレフィックスとサフィックスのフィルターで大文字と小文字が区別されるかどうかを示します

      イベントのフィルター処理の詳細については、Blob Storage のイベントに関するページをご覧ください。

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

      設定	推奨値	フィールドの説明
      他のデータベースへのデータのルーティングを許可する (マルチ データベース データ接続)	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 のデータ マッピングが自動生成されます。
      形式エラーを無視する	Ignore	JSON データ形式の形式エラーを無視する場合は、このオプションをオンに切り替えます。

      Note

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

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

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

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

      

5. [作成] を選択します

1. Azure portal でストレージ アカウントに移動します。
2. イベントを選択します。
3. [概要] タブで [Azure Data Explorer] タブを選択します。
4. [継続的なデータ インジェストの作成] タイルで、[作成] を選択します。

イベントをルーティング先として選択する

1. データの取り込み先となるクラスターを選択します。
2. ターゲット データベースを選択します。
3. [選択] を選択します。

データ接続を作成する

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

   

   設定	推奨値	フィールドの説明
   データ接続名	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 インジェスト用のリソースを手動で作成する」を参照してください。

   1. 必要に応じて、特定の Event Grid の件名を追跡できます。 次のように、通知用のフィルターを設定します。

      • [Prefix](プレフィックス) フィールドはサブジェクトの "リテラル" プレフィックスです。 適用されるパターンは [次の文字列で始まる] で、複数のコンテナー、フォルダー、BLOB を対象にできます。 ワイルドカードは使用できません。
        • BLOB コンテナーに対してフィルターを定義するには、フィールドを /blobServices/default/containers/[container prefix] のように設定する "必要があります"。
        • BLOB プレフィックス (または Azure Data Lake Gen2 のフォルダー) に対してフィルターを定義するには、フィールドを /blobServices/default/containers/[container name]/blobs/[folder/blob prefix] のように設定する "必要があります"。
      • [Suffix](サフィックス) フィールドは、BLOB の "リテラル" サフィックスです。 ワイルドカードは使用できません。
      • [Case-Sensitive](大文字と小文字の区別) フィールドは、プレフィックスとサフィックスのフィルターで大文字と小文字が区別されるかどうかを示します

      イベントのフィルター処理の詳細については、Blob Storage のイベントに関するページをご覧ください。

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

      設定	推奨値	フィールドの説明
      他のデータベースへのデータのルーティングを許可する (マルチ データベース データ接続)	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 のデータ マッピングが自動生成されます。
      形式エラーを無視する	Ignore	JSON データ形式の形式エラーを無視する場合は、このオプションをオンに切り替えます。

      Note

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

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

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

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

      

2. ［作成］ を選択します

次の例は、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')]"
            }
        }
    ]
}

次のサンプル コードでは、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 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);

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);