Managed Service for Apache Kafka クラスタの合成データを生成する
Google Cloud コンソールを使用して Google Cloud Managed Service for Apache Kafka クラスタの合成テストデータを生成する方法を学習します。
このガイドでは、Dataflow Streaming Data Generator テンプレートを使用して、サンプル ゲームのテレメトリー データを Managed Service for Apache Kafka トピックに自動的にパブリッシュします。ストリーミング データ生成ツールは、指定されたスキーマに基づいて、構成可能なレートで合成テストレコードを生成する Dataflow テンプレートです。合成データを生成すると、ローカル Kafka クライアントをインストールしたり、カスタム プロデューサー コードを記述したりすることなく、クラスタ アクティビティのモニタリング、負荷処理のテスト、モニタリング指標の検証を行うことができます。テンプレートの詳細については、Dataflow ストリーミング データ ジェネレータ テンプレートをご覧ください。
始める前に
このチュートリアルを開始する前に、新しい Managed Service for Apache Kafka クラスタを作成します。すでにクラスタがある場合は、この手順をスキップできます。クラスタの作成に必要なロールと権限については、クラスタを作成して表示するをご覧ください。そのガイドに沿って操作する場合は、このガイドに戻る前に、クラスタを作成するセクションのみを完了してください。
クラスタを作成する方法
コンソール
- [Managed Service for Apache Kafka] > [クラスタ] ページに移動します。
- [ 作成] をクリックします。
- [クラスタ名] フィールドに、クラスタの名前を入力します。
- [リージョン] リストで、クラスタのロケーションを選択します。
-
[ネットワーク構成] で、クラスタにアクセス可能なサブネットを構成します。
- [プロジェクト] で、該当するプロジェクトを選択します。
- [ネットワーク] で、VPC ネットワークを選択します。
- [サブネット] でサブネットを選択します。
- [完了] をクリックします。
- [作成] をクリックします。
[作成] をクリックすると、クラスタの状態が Creating になります。クラスタの準備が整うと、状態は Active になります。
gcloud
Kafka クラスタを作成するには、managed-kafka clusters
create コマンドを実行します。
gcloud managed-kafka clusters create KAFKA_CLUSTER \ --location=REGION \ --cpu=3 \ --memory=3GiB \ --subnets=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \ --async
次のように置き換えます。
KAFKA_CLUSTER: Kafka クラスタの名前REGION: クラスタのロケーションPROJECT_ID: プロジェクト IDSUBNET_NAME: クラスタを作成するサブネット（例:default）
サポートされているロケーションについては、 Managed Service for Apache Kafka のロケーションをご覧ください。
このコマンドは非同期で実行され、オペレーション ID を返します。
Check operation [projects/PROJECT_ID/locations/REGION/operations/OPERATION_ID] for status.
作成オペレーションの進行状況を追跡するには、gcloud managed-kafka
operations describe コマンドを使用します。
gcloud managed-kafka operations describe OPERATION_ID \ --location=REGION
クラスタの準備ができると、このコマンドの出力に state:
ACTIVE エントリが含まれます。詳細については、 クラスタ作成オペレーションをモニタリングするをご覧ください。
必要なロール
クラスタの合成データを生成するために必要な権限を取得するには、プロジェクトに対する次の IAM ロールを付与するよう管理者に依頼してください。
- Dataflow デベロッパー （
roles/dataflow.developer） - プロジェクト IAM 管理者 （
roles/resourcemanager.projectIamAdmin）
ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。
これらの事前定義ロールには、クラスタの合成データを生成するために必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。
必要な権限
クラスタの合成データを生成するには、次の権限が必要です。
-
dataflow.jobs.create -
dataflow.jobs.get -
managedkafka.clusters.get -
managedkafka.topics.get -
managedkafka.topics.create -
managedkafka.topics.publish -
resourcemanager.projects.setIamPolicy
カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。
Compute Engine のデフォルト サービス アカウントに Dataflow ジョブを実行するために必要な権限を付与するには、プロジェクトに対する次の IAM ロールを Compute Engine のデフォルト サービス アカウントに付与するよう管理者に依頼します。
- Dataflow ワーカー （
roles/dataflow.worker） - マネージド Kafka クライアント （
roles/managedkafka.client）
ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。
管理者は、カスタムロールや他の事前定義ロールを使用して、Compute Engine のデフォルトのサービス エージェントに必要な権限を付与することもできます。
IAM ロールを付与する権限がある場合、Google Cloud コンソールには、合成データ生成プロセス中に必要なロールを付与するよう求めるメッセージが表示されます。ロールを付与する権限がない場合、コンソールに、必要な権限を管理者に付与するようリクエストするメッセージが表示されます。
合成データを生成する
Kafka トピックの合成データを生成する Dataflow ジョブを作成して起動するには:
Google Cloud コンソールで、[Managed Service for Apache Kafka] > [クラスタ] ページに移動します。
クラスタの名前（
test-clusterなど）をクリックします。[Sources] タブを選択します。
[ソース] ページの [合成データを生成する] カードで、[Dataflow ジョブを作成する] をクリックします。[Produce Data] ペインが開きます。
[データの生成] ペインで、[Kafka トピック] プルダウン リストからトピック（
test-topicなど）を選択します。トピックがない場合は、次の手順で作成します。- [Kafka トピック] プルダウン リストで、[トピックを作成] をクリックします。[トピックを作成] ペインが開きます。
- [トピック名] フィールドに「
test-topic」と入力します。 - [パーティション数]（
3）と [レプリケーション係数]（3）のデフォルト値はそのままにします。 - [作成] をクリックします。
[出力レート（QPS）] フィールドに、ジェネレータで生成する秒間クエリ数（
100など）を入力します。これにより、クラスタがさまざまな負荷をどのように処理するかをテストできます。Dataflow サービス アカウントに必要な権限がないという警告が表示された場合は、[付与] をクリックして次のロールを割り当てます。
- Dataflow ワーカー（
roles/dataflow.worker） - マネージド Kafka クライアント（
roles/managedkafka.client）
- Dataflow ワーカー（
[データを生成] ペインで、[作成] をクリックして Dataflow ジョブを起動します。
Dataflow ジョブが作成されたことを示す通知が表示されます。
通知で [ジョブを表示] をクリックして [Dataflow ジョブの詳細] ページを開き、ジョブグラフ、ステータス、実行指標を確認します。
クラスタ指標を表示する
Dataflow ジョブが開始されたら、クラスタに流れる合成データを観察します。
test-clusterの [クラスタの詳細] ページで、[モニタリング] タブをクリックします。[バイトレート] グラフと [プロダクト スループット別の上位 5 つのトピック] グラフを確認して、データがトピックにアクティブに生成されていることを確認します。
メッセージを表示する
次のいずれかの方法で、合成メッセージがトピックにパブリッシュされていることを確認します。
Kafka コマンドライン ツールで表示する
クライアント VM で Kafka CLI ツールを使用してクラスタからメッセージを直接使用するには:
SSH を使用してクライアント VM に接続します。クライアント VM を設定していない場合は、クライアント VM を作成するをご覧ください。
Google Cloud コンソールからクラスタのブートストラップ サーバー アドレスを取得し、クライアント VM の環境変数として設定します。
Google Cloud コンソールで、[Managed Service for Apache Kafka] > [クラスタ] ページに移動します。
クラスタの名前（
test-clusterなど）をクリックします。[クラスタの詳細] ページで、[構成] をクリックします。
[ブートストラップ URL] の下にリストされている値をコピーします。
クライアント VM で、環境変数を設定します。
```sh export BOOTSTRAP="BOOTSTRAP_URL" ```BOOTSTRAP_URLは、コピーしたブートストラップ アドレスに置き換えます。kafka-console-consumer.shコマンドを実行してメッセージを読み取ります。kafka-console-consumer.sh \ --bootstrap-server $BOOTSTRAP \ --topic TOPIC_ID \ --from-beginning \ --consumer.config client.propertiesTOPIC_ID は、
test-topicなどのトピック名に置き換えます。コンソールに、ストリーミングされた合成ゲームデータ レコードが使用されるたびに表示されます。
Ctrl+C キーを押して、メッセージの消費を停止します。
BigQuery で表示
Kafka トピックから BigQuery にデータをストリーミングしてレコードを表示するには:
合成データは未加工の JSON であるため、コネクタを作成する前に、BigQuery で宛先テーブルを手動で作成する必要があります。テーブルの作成方法については、スキーマ定義を含む空のテーブルを作成するをご覧ください。次のスキーマを使用して、データセットに
test-topicという名前のテーブルを作成します。[ {"name": "eventId", "type": "STRING"}, {"name": "eventTimestamp", "type": "INTEGER"}, {"name": "ipv4", "type": "STRING"}, {"name": "ipv6", "type": "STRING"}, {"name": "country", "type": "STRING"}, {"name": "username", "type": "STRING"}, {"name": "quest", "type": "STRING"}, {"name": "score", "type": "INTEGER"}, {"name": "completed", "type": "BOOLEAN"} ]Connect クラスタに BigQuery Sink コネクタを作成して、トピックから BigQuery テーブルにメッセージをストリーミングします。コネクタを構成するときは、次のサンプル プロパティを使用します。
PROJECT_IDはプロジェクト ID に置き換えます。bigQueryPartitionDecorator=false connector.class=com.wepay.kafka.connect.bigquery.BigQuerySinkConnector defaultDataset=test_dataset key.converter=org.apache.kafka.connect.storage.StringConverter project=PROJECT_ID tasks.max=3 topics=test-topic value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=falseコネクタがデータのストリーミングを開始したら、 Google Cloud コンソールの [BigQuery] ページに移動します。
[エクスプローラ] パネルでプロジェクト ID を開き、データセット
test_datasetを選択します。テーブル名
test-topicをクリックします。[プレビュー] タブをクリックして、ストリーミングされた合成レコードを表示します。または、[クエリを新規作成] をクリックして、次の SQL クエリを実行します。
SELECT * FROM `PROJECT_ID.DATASET_ID.TABLE_ID` LIMIT 10;次のように置き換えます。
- PROJECT_ID: プロジェクト ID
- DATASET_ID: データセット ID（
test_datasetなど） - TABLE_ID: テーブル ID（
test-topicなど）
[実行] をクリックして、[クエリ結果] ペインにサンプルレコードを表示します。
注:
SELECT COUNT(*)クエリを使用してレコードを確認しないでください。コネクタは BigQuery Streaming API を使用するため、データは最初にストリーミング バッファに書き込まれます。SELECT *を使用するとデータはすぐに表示されますが、行数の更新には数分かかることがあります。
クリーンアップ
このページで使用したリソースについて、 Google Cloud アカウントに課金されないようにするには、次の手順を実施します。
Google Cloud コンソールで、[Dataflow ジョブ] ページに移動します。
トピック用に作成されたジョブの名前をクリックします。
[停止] をクリックします。
[キャンセル] を選択してから [ジョブの停止] をクリックします。
省略可: Kafka クラスタが不要になった場合は、[Managed Service for Apache Kafka クラスタ] ページに移動し、
test-clusterを選択して [削除] をクリックします。