Managed Service for Apache Kafka トピックから BigQuery テーブルに AVRO または JSON メッセージを読み込む必要がある場合は、Dataflow テンプレートを使用します。このドキュメントでは、 Google Cloud コンソールを使用して、このテンプレートに基づいてパイプラインを設定する方法について説明します。REST API または Google Cloud CLI を使用してこのパイプラインを設定することもできます。
Google Cloud 個の商品を使用
Kafka to BigQuery Dataflow テンプレートは、次の課金対象の Google Cloud プロダクトを使用します。料金計算ツールを使用すると、予想使用量に基づいて費用の見積もりを出すことができます。
- BigQuery: BigQuery は Google Cloudのサーバーレス データ ウェアハウスです。このソリューションでは、BigQuery が Kafka データの宛先として機能します。BigQuery Streaming Write API は、データを BigQuery に書き込むために使用されます。このデータは、BigQuery で SQL クエリを使用して分析したり、BigQuery ML で ML モデルを構築したり、ビジネス インテリジェンス アプリケーションを強化したりできます。
- Dataflow: Dataflow はフルマネージドのデータ処理サービスです。Kafka to BigQuery Dataflow は、Dataflow を利用して、Kafka トピックからデータを読み取り、必要な変換を実行して BigQuery に書き込むパイプラインを作成します。Dataflow の自動スケーリング機能と自己修復機能により、パイプラインが信頼性と効率性を維持して実行されます。
- Cloud Storage: コアデータフローに直接関与していませんが、Kafka メッセージに Avro バイナリ エンコードを使用している場合は、Avro スキーマ ファイルの保存に使用できます。これにより、Dataflow パイプラインはデータの構造を適切に解釈できます。
また、このソリューションでは Managed Service for Apache Kafka も使用します。
- Managed Service for Apache Kafka: Apache Kafka の実行に役立つ Google Cloud サービス。Managed Service for Apache Kafka を使用すると、インフラストラクチャの管理ではなく、イベント ドリブン システムとストリーミング データ パイプラインの構築に集中できます。Managed Service for Apache Kafka の料金の詳細については、料金ガイドをご覧ください。
始める前に
Kafka to BigQuery Dataflow テンプレートを起動する前に、次の操作が完了していることを確認してください。
Managed Service for Apache Kafka クラスタとトピックを作成します。
クラスタとトピックを作成する方法の 1 つは、Managed Service for Apache Kafka クイックスタートに沿って操作することです。
トピックに Avro レコードが含まれている場合は、追加のリソース要件について、Kafka メッセージ形式を構成するをご覧ください。
次の Google Cloud API を有効にします。
Dataflow
BigQuery
Cloud Storage
gcloud services enable dataflow.googleapis.com bigquery.googleapis.com \ storage.googleapis.comBigQuery データセットとテーブルを作成します。テーブルのスキーマが Kafka 入力トピックのスキーマと一致していることを確認します。
同じトピック内で複数のスキーマを使用していて、複数のテーブルに書き込む場合は、パイプラインを構成する前にテーブルを作成する必要はありません。
BigQuery データセットとテーブルの作成方法については、データセットを作成するとスキーマ定義を含む空のテーブルを作成するをご覧ください。
Dataflow ワーカー サービス アカウントにマネージド Kafka クライアント ロールを付与する
Dataflow ジョブを Managed Service for Apache Kafka に接続するには、Dataflow ワーカー サービス アカウントに特定の権限を付与する必要があります。このサービス アカウントは、Dataflow ジョブ内のすべてのワーカー VM に使用される ID であり、これらの VM から行われたリクエストはこのアカウントを使用します。
Kafka リソースへのアクセスを許可するには、Dataflow ワーカー サービス アカウントに roles/managedkafka.client ロールを付与する必要があります。このロールには、接続の確立に必要な managedkafka.clusters.connect 権限が含まれています。
ワーカー サービス アカウントの詳細については、Google Cloud のパイプラインのセキュリティと権限をご覧ください。
Dataflow サービス アカウントにマネージド Kafka クライアント ロールを付与する手順は次のとおりです。
コンソール
- Google Cloud コンソールで、[IAM] ページに移動します。
IAM に移動 - プロジェクトが、Managed Service for Apache Kafka クライアントがアクセスするコンシューマー プロジェクトに設定されていることを確認します。
- [アクセス権を付与] をクリックします。
- 新しいページの [プリンシパルの追加] に、使用している Dataflow ワーカー サービス アカウントのメールアドレスを入力します。
- [ロールを割り当てる] で、[マネージド Kafka クライアント] ロールを選択します。
- [保存] をクリックします。
gcloud CLI
-
Google Cloud コンソールで Cloud Shell をアクティブにします。
Google Cloud コンソールの下部で Cloud Shell セッションが開始し、コマンドライン プロンプトが表示されます。Cloud Shell はシェル環境です。Google Cloud CLI がすでにインストールされており、現在のプロジェクトの値もすでに設定されています。セッションが初期化されるまで数秒かかることがあります。
-
gcloud projects add-iam-policy-bindingコマンドを実行します。gcloud projects add-iam-policy-binding PROJECT_ID \ --member serviceAccount:SERVICE_ACCOUNT_EMAIL \ --role roles/managedkafka.client
次のように置き換えます。
-
PROJECT_ID は、プロジェクト ID です。
-
SERVICE_ACCOUNT_EMAIL は、Dataflow ワーカー サービス アカウントのメールアドレスです。
-
Kafka to BigQuery Dataflow テンプレートを起動する
コンソールのクラスタの詳細ページから、Kafka to BigQuery Dataflow テンプレートを起動できます。
-
Google Cloud コンソールで、[クラスタ] ページに移動します。
プロジェクトで作成したクラスタが一覧表示されます。
- クラスタの詳細ページを表示するには、クラスタ名をクリックします。
- クラスタの詳細ページの [トピック] で、任意のトピックの BigQuery Export のアイコンをクリックします。
[テンプレート「Kafka to BigQuery」を使用して Dataflow ジョブを作成する] ページが開きます。
以降のセクションの情報に従って、テンプレートのフィールドを構成します。
ジョブ名を入力する
[ジョブ名] フィールドに、Dataflow ジョブの名前を入力します。
名前は、プロジェクトで現在実行中のすべてのジョブ間で一意である必要があります。
パイプラインのリージョン エンドポイントを選択する
[リージョン エンドポイント] フィールドで、リージョン エンドポイントを Kafka クラスタまたは BigQuery データセットのロケーションに設定して、リージョン間のデータ転送料金を最小限に抑えます。
Dataflow ワーカーは、Kafka クラスタのリージョンとは関係なく実行できます。ただし、Kafka クラスタのリージョン外でワーカーを起動すると、リージョン間の下り（外向き）料金が発生します。
クラスタのロケーションを表示するには、Managed Service for Apache Kafka クラスタを一覧表示するの手順に沿って操作します。
Dataflow テンプレートを選択する
[Dataflow テンプレート] で、デフォルト値の [Kafka to BigQuery] をそのままにします。
ソースの構成
[ソース] で、デフォルト値の [Managed Service for Apache Kafka] を保持します。
[Kafka クラスタ]、[Kafka トピック]、[Kafka ソース認証モード] のデフォルト値は、選択したトピックに基づいてすでに選択されています。これらの値を保持します。
Kafka メッセージ形式を構成する
Dataflow テンプレートは、次の 3 つのメッセージ形式をサポートしています。
Avro Confluent ワイヤー形式: 各 Kafka メッセージには、マジック バイト、スキーマ ID、Avro バイナリ エンコード レコードが含まれます。デフォルトでは、メッセージキーは BigQuery に読み込まれません。
Avro（Confluent ワイヤー形式）形式の場合、単一のスキーマまたは複数のスキーマを使用できます。
単一スキーマ: すべてのメッセージは、事前に定義された単一の Avro スキーマに準拠します。
複数のスキーマ: メッセージは異なるスキーマを利用できます。これは、Avro（Confluent ワイヤー形式）でのみサポートされています。
Avro（バイナリ エンコード）: メッセージには、メタデータを含まないレコードのペイロードのみが含まれます。Cloud Storage にアップロードされた Avro スキーマ ファイル（.avsc）を指定する必要があります。すべてのメッセージはこの単一のスキーマに準拠する必要があります。
JSON: レコードに事前定義されたスキーマは必要ありません。BigQuery テーブル スキーマに準拠していないレコードは、デッドレター キュー（構成されている場合）に送信されるか、エラー メッセージがロギングされます。サポートされている形式は
{"field": "value"}形式です。形式[{"name": "field", "value": "value"}]はサポートされていません。
Avro Confluent ワイヤー形式
[Kafka message format] でこのオプションを選択した場合は、次の追加設定を行います。
スキーマソース: このフィールドは、スキーマの場所をパイプラインに伝えます。次のいずれかのオプションを選択します。
スキーマ レジストリ: スキーマは Confluent Schema Registry に保存されます。これは、スキーマの進化と複数のバージョンの管理に役立ちます。スキーマ レジストリが Managed Service for Apache Kafka クラスタ ネットワークからアクセス可能であり、Dataflow ワーカーと同じリージョンでホストされていることを確認します。スキーマ レジストリは、単一スキーマと複数スキーマの両方のシナリオで使用できます。次の追加設定を行います。
スキーマ レジストリ接続 URL: スキーマ レジストリに接続するための URL を指定します。
認証モード: レジストリで認証が必要な場合は、[OAuth] または [TLS] を選択します。それ以外の場合は、[なし] を選択します。
単一のスキーマ ファイル: すべてのメッセージがファイルで定義された単一の固定スキーマに従う場合は、このオプションを選択します。
- Avro スキーマ ファイルの Cloud Storage ファイル: トピック内のすべてのメッセージをデコードするために使用される Avro スキーマ ファイルのパス。
Managed Service for Apache Kafka にはスキーマ レジストリがありません。このテンプレートは、Confluent ワイヤー形式と互換性のあるスキーマ レジストリへの認証情報の受け渡しのみをサポートしています。
Avro バイナリ エンコード
[Kafka message format] でこのオプションを選択した場合は、次の追加設定を行います。
- Avro スキーマ ファイルの Cloud Storage ファイル: トピック内のすべてのメッセージをデコードするために使用される Avro スキーマ ファイルのパス。
JSON
このオプションを [Kafka message format] として選択した場合、他の構成は必要ありません。
Kafka キーを BigQuery テーブルに保持するように構成する
[Kafka メッセージキーを BigQuery テーブルに保持する] オプションを有効にする場合は、ターゲット テーブルに BYTES 型の _key という名前のフィールドを追加する必要があります。キーは構造化データを表す場合がありますが、テンプレートではバイト配列として扱われます。
オフセット オプションを指定する
個々のワーカーまたはパイプライン全体を再起動するときにメッセージの再処理を回避するには、[Commit offsets to Kafka] オプションを選択します。これにより、パイプラインは中断したところから処理を再開し、重複処理やデータの不整合を防ぐことができます。
[Enter Consumer Group ID] フィールドに、このパイプラインのグループの一意の名前を入力します。ほとんどの場合、パイプラインは各メッセージを 1 回読み取り、再起動可能にする必要があります。
[デフォルトの Kafka 開始オフセット] フィールドでは、Dataflow パイプラインに 2 つの開始オフセット オプションが用意されています。次のいずれかを選択します。
Earliest: Kafka トピックの先頭からメッセージを処理します。
最新: 使用可能な最新のオフセットからメッセージの処理を開始します。
宛先の構成
これらのオプションは、データ パイプラインが BigQuery にデータを書き込む方法を制御します。
[テーブル名の戦略] フィールドでは、BigQuery テーブルにデータを整理する方法を指定します。これには次の 2 つの選択肢があります。
単一のテーブル名（デフォルト）: Kafka トピックのすべてのデータが 1 つの BigQuery テーブルに格納されます。これは最もシンプルなオプションであり、Kafka データのスキーマが同じ場合に推奨されるオプションです。
[BigQuery 出力テーブル] に、Kafka トピックのすべてのデータを保存するテーブル名を指定します。
動的テーブル名: Kafka メッセージのスキーマに基づいて複数の BigQuery テーブルを作成します。このオプションは、Kafka メッセージのスキーマが異なる場合に推奨されます。このオプションは、パイプラインでメッセージ構造の処理に Avro Confluent Wire またはスキーマ レジストリを使用している場合にのみ機能します。
Kafka メッセージの一意のスキーマごとに、BigQuery に独自のテーブルが作成されます。テーブル名は、スキーマ名に基づいて自動的に生成されます。
[動的テーブル名] を選択した場合は、次の追加フィールドを構成します。
BigQuery 出力プロジェクト: BigQuery データセットが存在する Google Cloud プロジェクト。
BigQuery 出力データセット: テーブルが作成されるプロジェクト内のデータセット。
BigQuery テーブル名の接頭辞（省略可）: 自動生成されるテーブル名に接頭辞を追加して、整理しやすくします。
BigQuery Storage Write API を構成する
[BigQuery Storage Write API のストリーム数] は、0（デフォルト）から始めます。このオプションを使用すると、パイプラインで最適なストリーム数を自動的に決定できます。
この設定は、BigQuery にデータを書き込むために使用される並列ストリームの数を制御します。数値を大きくすると、特に大量のデータの場合に書き込み速度が向上します。最適な数は、データと BigQuery の設定によって異なります。
トリガーの頻度を構成する
[BigQuery Write API のトリガー頻度（秒単位）] オプションでは、5 秒（デフォルト）から始めます。これは、ほとんどのユースケースでバランスの取れた設定です。
この設定により、BigQuery にデータが書き込まれる頻度が決まります。
数字が小さいほど、書き込み頻度が高くなります。これによりレイテンシを短縮できますが、BigQuery の費用が増加する可能性があります。数値が大きいほど、書き込みの頻度が低くなります。費用対効果は高くなりますが、レイテンシが増加する可能性があります。
デッドレター キューを構成する
破損、互換性のないデータ型、宛先 BigQuery テーブルとのスキーマの不一致が原因で、メッセージを処理できないことがあります。
このようなケースを処理するには、テンプレートでデッドレター キューを有効にして、テーブル名を指定します。テンプレートは、標準化されたスキーマを使用してテーブルを作成します。エラー メッセージは別の BigQuery テーブルに書き込まれます。
暗号化を構成する
デフォルトでは、すべての保存データと転送データはGoogle-owned and Google-managed encryption keyによって暗号化されます。顧客管理の暗号鍵（CMEK）がある場合は、独自の鍵を選択できます。CMEK の構成方法の詳細については、メッセージ暗号化を構成するをご覧ください。
ネットワークを構成する
Dataflow テンプレートで、クラスタのネットワークとサブネットワークを指定する必要があります。テンプレートの [省略可能なパラメータ] セクションでは、Dataflow ワーカーのネットワークを定義できます。
Kafka to BigQuery Dataflow テンプレートは、デフォルトで、プロジェクトのデフォルト ネットワークに Dataflow ワーカーをプロビジョニングします。Managed Service for Apache Kafka クラスタが Dataflow を介して BigQuery にデータを送信できるようにするには、Dataflow ワーカーがクラスタのネットワークにアクセスできることを確認します。
Kafka クラスタがプロジェクトのデフォルト ネットワークのサブネットに接続されていない場合は、Kafka クラスタにプロジェクトのデフォルト ネットワークを使用することをおすすめします。
Dataflow パイプラインでネットワークを設定する方法については、以下をご覧ください。
Dataflow ネットワーキングの構成で問題が発生した場合は、Dataflow ネットワーキングのトラブルシューティング ガイドをご覧ください。
オプションの Dataflow パラメータを構成する
省略可能なパラメータは、構成が Dataflow ワーカーに与える影響を理解している場合にのみ構成してください。設定が正しくないと、パフォーマンスや費用に影響する可能性があります。各オプションの詳細については、オプションのパラメータをご覧ください。
モニタリング
Kafka to BigQuery 用の Dataflow テンプレートは、コンソール内でログ、指標、エラーを探索できるモニタリング エクスペリエンスを提供します。このモニタリング ツールスイートは、Dataflow ユーザー インターフェースの一部として利用できます。
[ジョブ指標] タブでは、カスタム ダッシュボードを作成できます。Kafka から BigQuery へのテンプレートでは、次の項目をモニタリングするジョブ指標ダッシュボードを設定することをおすすめします。
スループット: 任意の時点で処理されたデータ量。これは、ジョブのデータフローをモニタリングし、パフォーマンスの問題を特定するのに役立ちます。
詳細については、Dataflow スループットのモニタリングをご覧ください。
データの更新速度: データ要素のタイムスタンプとパイプラインでイベントが処理される時間との差（秒単位）。これにより、パフォーマンスとデータソースのボトルネックや頻繁な再試行を特定できます。
詳細については、Dataflow データの更新速度モニタリングをご覧ください。
バックログ: 処理待ちのバイト数。この情報は、自動スケーリングの決定に役立ちます。
Dataflow モニタリングの詳細については、Dataflow モニタリングのドキュメントをご覧ください。
トラブルシューティング
Dataflow パイプラインでパフォーマンスの問題が発生した場合は、Dataflow に用意されている包括的なトラブルシューティング ツールと診断ツールを使用できます。
一般的なシナリオと、それぞれのトラブルシューティング ガイドは次のとおりです。
Dataflow パイプラインのデバッグの概要については、Dataflow パイプラインのトラブルシューティングとデバッグをご覧ください。
既知の制限事項
このテンプレートでは、認証用の認証情報をスキーマ レジストリに渡すことはできません。
Kafka から BigQuery への Dataflow ジョブを作成するときに、 Google Cloud プロジェクトが Managed Service for Apache Kafka クラスタを含む同じプロジェクトに設定されていることを確認します。