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

Cribl Stream は、あらゆるソースからマシン イベント データを安全に収集、処理、ストリームする処理エンジンです。 これにより安全な方法で、分析と管理を目的として、あらゆる宛先に対してそのデータを解析および処理できます。

この記事では、Cribl Stream を使用してデータを取り込む方法について説明します。

データ コネクタの完全な一覧については、「データ統合の概要」を参照してください。

前提条件

• Cribl Stream アカウント
• 既定のキャッシュおよび保持ポリシーが設定された Azure Data Explorer クラスターおよびデータベース。
• クエリ環境。 詳細については、「クエリ統合の概要」を参照してください。
• "https://ingest-<cluster>.<region>.kusto.windows.net" のフォーマットで "TargetURI" 値に Kusto クラスター URI を指定します。 詳細については、「クラスター接続の追加」を参照してください。

Microsoft Entra サービス プリンシパルを作成する

Microsoft Entra サービス プリンシパルは、次の例のように Azure portal を通してか、プログラムを使用して作成できます。

このサービス プリンシパルは、Kusto 内のテーブルにデータを書き込むコネクタによって使用される ID です。 Kusto リソースにアクセスするためのアクセス許可をこのサービス プリンシパルに付与します。

1. Azure CLI 経由で Azure サブスクリプションにサインインします。 次に、ブラウザーで認証します。

   az login
   

2. プリンシパルをホストするサブスクリプションを選択します。 この手順は、複数のサブスクリプションがある場合に必要です。

   az account set --subscription YOUR_SUBSCRIPTION_GUID
   

3. サービス プリンシパルを作成します。 この例では、サービス プリンシパルを my-service-principal と呼びます。

   az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}
   

4. 返された JSON データから、appId、password、および tenant を後で使用のためにコピーします。

   {
     "appId": "1234abcd-e5f6-g7h8-i9j0-1234kl5678mn",
     "displayName": "my-service-principal",
     "name": "my-service-principal",
     "password": "1234abcd-e5f6-g7h8-i9j0-1234kl5678mn",
     "tenant": "1234abcd-e5f6-g7h8-i9j0-1234kl5678mn"
   }
   

Microsoft Entra アプリケーションとサービス プリンシパルが作成されました。

ターゲット テーブルを作成する

受信データのターゲット テーブルと、取り込まれたデータ列をターゲット テーブル内の列にマップするためのインジェスト マッピングを作成します。

1. プレースホルダー TableName をターゲット テーブルの名前に置き換えて、クエリ エディターで次のテーブル作成コマンドを実行します。

   .create table <TableName> (_raw: string, _time: long, cribl_pipe: dynamic)
   

2. プレースホルダーの TableName をターゲット テーブル名に、TableNameMapping をインジェスト マッピング名に置き換えて、次の create ingestion mapping コマンドを実行します。

   .create table <TableName> ingestion csv mapping '<TableNameMapping>' 'CriblLogMapping' '[{"Name":"_raw","DataType":"string","Ordinal":"0","ConstValue":null},{"Name":"_time","DataType":"long","Ordinal":"1","ConstValue":null},{"Name":"cribl_pipe","DataType":"dynamic","Ordinal":"2","ConstValue":null}]'
   

3. 「Microsoft Entra サービス プリンシパルの作成」からサービス プリンシパルにデータベースを操作するためのデータベース インジェスター ロール アクセス許可を付与します。 詳細については、「例」を参照してください。 プレースホルダー DatabaseName をターゲット データベースの名前に、ApplicationID を Microsoft Entra サービス プリンシパルの作成時に保存した AppId 値に置き換えます。

   .add database <DatabaseName> ingestors ('aadapp=<ApplicationID>') 'App Registration'
   

Cribl Stream の宛先を作成する

次のセクションでは、Kusto 内のテーブルにデータを書き込む Cribl Stream の宛先を作成する方法について説明します。 各テーブルには、個別の Cribl Stream 宛先コネクタが必要です。

宛先の選択

Cribl Stream をテーブルに接続するには:

1. Cribl 内の上部のナビゲーションで、[Manage] を選択してから [Worker Group] を選択します。

2. [Routing]>[QuickConnect (Stream)]>[Add Destination] を選択します。

3. [Set up new QuickConnect Destination] ウィンドウ内で、[Azure Data Explorer]、次に [Add now] を選択します。

General Settings を設定する

[New Data Explorer] ウィンドウ内の、[General Settings] 内で次の設定を行います。

完了したら [Next] を選択します。

認証設定

サイドバー内で [Authentication settings] を選択します。 「Microsoft Entra サービス プリンシパルを作成する」の中で保存した値とベース URI を次のように使用します。

[次へ] を選択します。

Persistent Queue

Ingestion Mode が Streaming に設定され、かつ Backpressure behavior が Persistent Queue に設定されている場合に表示されます。

完了したら、[次へ] を選択します。

Processing settings

完了したら、[次へ] を選択します。

Parquet settings

Data Format に Parquet が選択されている場合に表示されます。

Parquet を選択すると [Parquet Settings] タブが開き、Parquet スキーマを選択できます。

再試行

Ingestion Mode が Streaming に設定されている場合に表示されます。

完了したら、[次へ] を選択します。

詳細設定

サイドバーから [Advanced Settings] を選択します。 次に、Batching が選択されている場合の詳細設定について説明します。

完了したら [Save] を選択します。

接続の構成

開いた [Connection Configuration] ウィンドウで、[Passthru connection] と [Save] を選択します。 コネクタはデータのエンキューを開始します。

データ インジェストを確認する

1. テーブル内にデータが到着したら、行数をチェックすることでデータの転送を確認します。

   <Tablename> 
   | count
   

2. 直近の 5 分間にエンキューされたインジェストを確認します。

   .show commands-and-queries 
   | where Database == "" and CommandType == "DataIngestPull" 
   | where LastUpdatedOn >= ago(5m)
   

3. インジェスト プロセスでエラーが発生していないことを確認します。

   • Batching の場合:

   .show ingestion failures
   

   • Streaming の場合:

   .show streamingingestion failures 
   | order by LastFailureOn desc
   

4. テーブル内のデータを確認します。

   <TableName>
   | take 10
   

クエリの例とガイダンスについては、KQL でクエリを記述すると Kusto 照会言語のドキュメントを参照してください。

関連するコンテンツ

• データ統合の概要
• Kusto クエリ言語 (KQL) の概要