Apache Flink を使用してデータを Azure Data Explorer に取り込む

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

Apache Flink は、境界なしと境界ありのデータストリームに対するステートフルな計算のためのフレームワークおよび分散処理エンジンです。

Flink コネクタは、あらゆる Flink クラスターで実行できるオープンソースプロジェクトです。 Flink クラスターからデータを移動するためのデータシンクを実装します。 Apache Flink へのコネクタを使用すると、機械学習 (ML)、抽出・変換・読み込み (ETL)、Log Analytics などのデータドリブンシナリオを対象とする高速でスケーラブルなアプリケーションを構築できます。

この記事では、Flink コネクタを使用して Flink からテーブルにデータを送信する方法について説明します。テーブルとデータマッピングを作成し、Flink にテーブルにデータを送信するように指示し、結果を検証します。

前提条件

Azure Data Explorer クラスターとデータベース。クラスターとデータベースを作成するか、または Microsoft Fabric のリアルタイムインテリジェンスで KQL データベースを作成します。
データベース内のターゲットテーブル。「Azure Data Explorer でテーブルを作成する」または「Real-Time Intelligence でテーブルを作成する」を参照してください
Apache Flink クラスター。クラスターを作成します。
Maven 3.x

Flink コネクタを取得する

Maven を使用して依存関係を管理する Flink プロジェクトの場合は、依存関係として追加して、Flink Connector Core Sink for Azure Data エクスプローラーを統合します。

<dependency>
    <groupId>com.microsoft.azure.kusto</groupId>
    <artifactId>flink-connector-kusto</artifactId>
    <version>1.0.0</version>
</dependency>

依存関係の管理に Maven を使用しないプロジェクトの場合は、Azure Data Explorer Connector for Apache Flink のリポジトリを複製し、ローカルにビルドします。この方法では、コマンド mvn clean install -DskipTestsを使用してローカル Maven リポジトリにコネクタを手動で追加できます。

Application
マネージド ID

Microsoft Entra ID アプリケーションまたはマネージド ID のいずれかを使用して Flink から認証できます。

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

Azure CLI 経由で Azure サブスクリプションにサインインします。次に、ブラウザーで認証します。
```
az login
```
プリンシパルをホストするサブスクリプションを選択します。この手順は、複数のサブスクリプションがある場合に必要です。
```
az account set --subscription YOUR_SUBSCRIPTION_GUID
```
サービスプリンシパルを作成します。この例では、サービスプリンシパルを my-service-principal と呼びます。
```
az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}
```

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

{
  "appId": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "displayName": "my-service-principal",
  "name": "my-service-principal",
  "password": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "tenant": "00001111-aaaa-2222-bbbb-3333cccc4444"
}

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

データベースに対するアプリケーションユーザー権限を付与します。

// Grant database user permissions
.add database <MyDatabase> users ('aadapp=<Application ID>;<Tenant ID>')

アプリケーションに、テーブルに対するインジェストまたは管理者のアクセス許可を付与します。必要な権限は、選択したデータ書き込み方法によって異なります。 SinkV2 には取り込み権限で十分ですが、WriteAndSink には管理者アクセス許可が必要です。
```
// Grant table ingestor permissions (SinkV2)
.add table <MyTable> ingestors ('aadapp=<Application ID>;<Tenant ID>')

// Grant table admin permissions (WriteAheadSink)
.add table <MyTable> admins ('aadapp=<Application ID>;<Tenant ID>')
```

詳細については、「Kusto ロールベースのアクセスコントロール」を参照してください。

マネージド ID 認証を使用するには:

システム割り当てマネージド ID またはユーザー割り当てマネージド ID をクラスターに追加します。 オブジェクト ID を保存します。
マネージド ID に Azure Data Explorer データベースでのユーザー権限を付与します。
```
// Grant database user permissions
.add database <MyDatabase> users ('aadapp=<Object ID>;<Tenant ID>')
```
マネージド ID に Azure Data Explorer テーブルでの取り込み者権限または管理者権限を付与します。必要な権限は、選択したデータ書き込み方法によって異なります。 SinkV2 には取り込み権限で十分ですが、WriteAndSink には管理者アクセス許可が必要です。
```
// Grant table ingestor permissions (SinkV2)
.add table <MyTable> ingestors ('aadapp=<Object ID>;<Tenant ID>')

// Grant table admin permissions (WriteAheadSink)
.add table <MyTable> admins ('aadapp=<Object ID>;<Tenant ID>')
```

詳細については、「Kusto ロールベースのアクセスコントロール」を参照してください。

Flink からデータを書き込む

Flink からデータを書き込むには：

必要な関数をインポートするには：

import com.microsoft.azure.flink.config.KustoConnectionOptions;
import com.microsoft.azure.flink.config.KustoWriteOptions;

アプリケーションまたはマネージド ID を使用して認証します。

アプリケーションによる認証の場合:

KustoConnectionOptions kustoConnectionOptions = KustoConnectionOptions.builder()
.setAppId("<Application ID>")
.setAppKey("<Application key>")
.setTenantId("<Tenant ID>")
.setClusterUrl("<Cluster URI>").build();

マネージド ID による認証の場合:

KustoConnectionOptions kustoConnectionOptions = KustoConnectionOptions.builder()
.setManagedIdentityAppId("<Object ID>")
.setClusterUrl("<Cluster URI>").build();

データベースやテーブルなどのシンクパラメーターを構成します。

KustoWriteOptions kustoWriteOptions = KustoWriteOptions.builder()
    .withDatabase("<Database name>").withTable("<Table name>").build();

You can add more options, as described in the following table:

オプション	説明	Default Value
IngestionMappingRef	既存のインジェストマッピングを参照します。
FlushImmediately	データをすぐにフラッシュし、パフォーマンスの問題を引き起こす可能性があります。この方法は推奨されていません。
BatchIntervalMs	データをフラッシュする頻度をコントロールします。	30 秒
BatchSize	フラッシュ前にレコードをバッファリングするためのバッチサイズを設定します。	1,000 レコード
ClientBatchSizeLimit	インジェスト前の集計データのMBサイズを指定します。	300 MB
PollForIngestionStatus	true の場合、コネクタはデータフラッシュ後にインジェスト状態をポーリングします。	false
DeliveryGuarantee	配信保証のセマンティクスを決定します。 1 回だけセマンティクスを実現するには、WriteAheadSink を使用します。	AT_LEAST_ONCE

次のいずれかの方法でストリーミングデータを書き込みます。
- SinkV2: これは、少なくとも一度の整合性を確保して、チェックポイントでデータをフラッシュするステートレスオプションです。大量のデータインジェストには、このオプションをお勧めします。
- WriteAheadSink: このメソッドは、KustoSink にデータを出力します。 Flinkのチェックポイントシステムと統合されており、一度回だけ保証を提供します。データは AbstractStateBackend に記憶され、チェックポイントが完了した後にのみコミットされます。
次の例では、SinkV2 を使用します。 WriteAheadSink を使用するには、buildWriteAheadSinkの代わりに方法buildを使用します。
```
KustoWriteSink.builder().setWriteOptions(kustoWriteOptions)
    .setConnectionOptions(kustoConnectionOptions).build("<Flink source datastream>" /*Flink source data stream, example messages de-queued from Kafka*/
    , 2 /*Parallelism to use*/);
```

完全なコードは次のようになります。

import com.microsoft.azure.flink.config.KustoConnectionOptions;
import com.microsoft.azure.flink.config.KustoWriteOptions;

KustoConnectionOptions kustoConnectionOptions = KustoConnectionOptions.builder()
.setAppId("<Application ID>")
.setAppKey("<Application key>")
.setTenantId("<Tenant ID>")
.setClusterUrl("<Cluster URI>").build();

KustoWriteOptions kustoWriteOptions = KustoWriteOptions.builder()
    .withDatabase("<Database name>").withTable("<Table name>").build();

KustoWriteSink.builder().setWriteOptions(kustoWriteOptions)
    .setConnectionOptions(kustoConnectionOptions).build("<Flink source datastream>" /*Flink source data stream, example messages de-queued from Kafka*/
    , 2 /*Parallelism to use*/);

データが取り込まれていることを確認する

接続が構成されると、データがテーブルに送信されます。データが取り込まれていることは、KQL クエリを実行することで確認できます。

次のクエリを実行して、データがテーブルに取り込まれていることを確認します。
```
<MyTable>
| count
```
Run the following query to view the data:
```
<MyTable>
| take 100
```

次の方法で共有

Apache Flink を使用してデータを Azure Data Explorer に取り込む

前提条件

Flink コネクタを取得する

Flink からデータを書き込む

データが取り込まれていることを確認する

フィードバック

その他のリソース

次の方法で共有

Apache Flink を使用してデータを Azure Data Explorer に取り込む

前提条件

Flink コネクタを取得する

Flink からデータを書き込む

データが取り込まれていることを確認する

関連するコンテンツ

フィードバック

その他のリソース