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

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

2. [データ] セクションで [データベース]>[TestDatabase] を選択します。

   

3. [設定] セクションで [データ接続] を選択した後、[データ接続の追加]>[Event Grid (BLOB ストレージ)] を選択します。

   

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

   

   設定	推奨値	フィールドの説明
   データ接続名	テストグリッド接続	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 のサブジェクトを追跡できます。 次のように、通知用のフィルターを設定します。

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

      イベントのフィルター処理の詳細については、Blob Storage のイベントに関する記事を参照してください。

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

      設定	推奨値	フィールドの説明
      他のデータベースへのデータのルーティングを許可する (マルチ データベース データ接続)	許可しないでください	このオプションは、データ接続に関連付けられている既定のターゲット データベースをオーバーライドする場合にオンにします。 データベースのルーティングの詳細については、「イベント ルーティング」を参照してください。
      テーブル名	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 データの形式エラーを無視する場合は、このオプションをオンにします。

      注

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

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

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

      [なし] を選択すると、ストレージ アカウントと Event Hub は接続文字列によって認証されます。 この方法はお勧めできません。

      

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

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

イベントのルーティング先を選択します

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

データ接続を作成する

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

   

   設定	推奨値	フィールドの説明
   データ接続名	テストグリッド接続	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 のサブジェクトを追跡できます。 次のように、通知用のフィルターを設定します。

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

      イベントのフィルター処理の詳細については、Blob Storage のイベントに関する記事を参照してください。

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

      設定	推奨値	フィールドの説明
      他のデータベースへのデータのルーティングを許可する (マルチ データベース データ接続)	許可しないでください	このオプションは、データ接続に関連付けられている既定のターゲット データベースをオーバーライドする場合にオンにします。 データベースのルーティングの詳細については、「イベント ルーティング」を参照してください。
      テーブル名	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 データの形式エラーを無視する場合は、このオプションをオンにします。

      注

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

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

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

      [なし] を選択すると、ストレージ アカウントと Event Hub は接続文字列によって認証されます。 この方法はお勧めできません。

      

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