Google Cloud Managed Service for Apache Kafka トピックから Cloud Storage バケットにデータを読み込む必要がある場合は、Dataflow テンプレートを使用します。 Google Cloud コンソール、REST API、または Google Cloud CLI を使用できます。
このドキュメントでは、 Google Cloud コンソールを使用して Kafka to Cloud Storage Dataflow テンプレートを構成する方法について説明します。
Google Cloud 個の商品が使用されました
Kafka to Cloud Storage Dataflow テンプレートは、次の課金対象の Google Cloud プロダクトを使用します。料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを出すことができます。
- Dataflow: Dataflow はフルマネージドのデータ処理サービスです。Kafka to Cloud Storage Dataflow テンプレートは、Dataflow を使用して、Kafka トピックからデータを読み取り、必要な変換を実行して、Cloud Storage に書き込むパイプラインを作成します。Dataflow の自動スケーリング機能と自己修復機能により、パイプラインが信頼性と効率性を維持して実行されます。
- Cloud Storage: Kafka データの宛先として機能します。Dataflow パイプラインによって転送されたデータを保存するには、Cloud Storage バケットが必要です。
また、このソリューションでは Google Cloud Managed Service for Apache Kafka も使用します。
- Google Cloud Managed Service for Apache Kafka: Apache Kafka の実行を支援する Google Cloud サービス。パイプラインのソースデータを提供します。Cloud Storage に転送するデータを含む既存の Managed Service for Apache Kafka クラスタとトピックが必要です。Google Cloud Managed Service for Apache Kafka の料金の詳細については、料金ガイドをご覧ください。
始める前に
Kafka から Cloud Storage への Dataflow テンプレートを起動する前に、次のことを完了していることを確認してください。
Managed Service for Apache Kafka クラスタとトピックを作成します。
クラスタとトピックを作成する方法の 1 つは、Managed Service for Apache Kafka クイックスタートに沿って操作することです。
トピックに Avro レコードが含まれている場合は、追加のリソース要件について、メッセージ形式を指定するをご覧ください。
次の Google Cloud API を有効にします。
Dataflow
Cloud Storage
gcloud services enable dataflow.googleapis.com storage-api.googleapis.com \Cloud Storage バケットを作成する。
Cloud Storage バケットの作成方法の詳細については、バケットを作成するをご覧ください。
マネージド Kafka クライアント ロールを Dataflow ワーカー サービス アカウントに付与する
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 Cloud Storage Dataflow テンプレートを起動する
コンソールのクラスタの詳細ページから Kafka to Cloud Storage Dataflow テンプレートを起動できます。
-
Google Cloud コンソールで、[クラスタ] ページに移動します。
プロジェクトで作成したクラスタが一覧表示されます。
- クラスタの詳細ページを表示するには、クラスタ名をクリックします。
- クラスタの詳細ページで、[データをインポート] をクリックします。
[テンプレート「Kafka to Kafka」を使用して Dataflow ジョブを作成する] ページが開きます。
- テンプレートの [Dataflow テンプレート] で、テンプレートを [Kafka to Cloud Storage] に更新します。
以降のセクションで説明する情報に基づいて、テンプレートのフィールドを構成します。
ジョブ名を入力する
[ジョブ名] フィールドに、Dataflow ジョブの名前を入力します。
名前は、プロジェクトで現在実行中のすべてのジョブの中で一意である必要があります。
パイプラインのリージョン エンドポイントを選択する
[リージョン エンドポイント] フィールドで、リージョン エンドポイントを Kafka クラスタのロケーションに設定して、リージョン間のデータ移転料金を最小限に抑えます。
Dataflow ワーカーは、Kafka クラスタのリージョンとは独立して実行できます。ただし、Kafka クラスタのリージョン外でワーカーを起動すると、リージョン間の下り（外向き）費用が発生します。
クラスタのロケーションを表示するには、Managed Service for Apache Kafka クラスタを一覧表示するの手順に沿って操作します。
ソースの構成
[Source] で、デフォルト値の [Managed Service for Apache Kafka] を保持します。
[Kafka クラスタ] と [Kafka ソース認証モード] はデフォルト値のままにします。
[Kafka トピック] で、使用可能なトピックのリストからトピックを選択します。
Kafka メッセージ形式を構成する
Dataflow テンプレートは、次の 3 つのメッセージ形式をサポートしています。
Avro Confluent ワイヤー形式: 各 Kafka メッセージには、マジック バイト、スキーマ ID、Avro バイナリ エンコード レコードが含まれます。
Avro（Confluent ワイヤー形式）形式の場合、単一のスキーマまたは複数のスキーマを使用できます。
単一スキーマ: すべてのメッセージは、事前に定義された単一の Avro スキーマに準拠します。
複数のスキーマ: メッセージは異なるスキーマを利用できます。これは、Avro（Confluent ワイヤー形式）でのみサポートされています。
Avro（バイナリ エンコード）: メッセージには、メタデータを含まないレコードのペイロードのみが含まれます。Cloud Storage にアップロードされた Avro スキーマ ファイル（.avsc）を指定する必要があります。すべてのメッセージはこの単一のスキーマに準拠する必要があります。
JSON: レコードに事前定義されたスキーマは必要ありません。スキーマに準拠していないレコードは、デッドレター キュー（構成されている場合）に送信されるか、エラー メッセージがログに記録されます。サポートされている形式は
{"field": "value"}形式です。フォーマット[{"name": "field", "value": "value"}]はサポートされていません。
Google Cloud Managed Service for Apache Kafka にはスキーマ レジストリがありません。このテンプレートは、Confluent ワイヤー形式と互換性のあるスキーマ レジストリへの認証情報の受け渡しのみをサポートしています。
Avro Confluent ワイヤー形式
このオプションを Kafka メッセージ形式として選択した場合は、次の追加設定を行います。
スキーマ ソース: このフィールドは、スキーマの検索場所をパイプラインに伝えます。次のいずれかのオプションを選択します。
スキーマ レジストリ: スキーマは Confluent Schema Registry に保存されます。これは、スキーマの進化と複数のバージョンの管理に役立ちます。スキーマ レジストリが Managed Service for Apache Kafka クラスタ ネットワークからアクセス可能であり、Dataflow ワーカーと同じリージョンでホストされていることを確認します。スキーマ レジストリは、単一スキーマと複数スキーマの両方のシナリオで使用できます。次の追加設定を行います。
スキーマ レジストリ接続 URL: スキーマ レジストリに接続するための URL を指定します。
認証モード: レジストリで認証が必要な場合は、[OAuth] または [TLS] を選択します。それ以外の場合は、[なし] を選択します。
単一のスキーマ ファイル: すべてのメッセージがファイルで定義された単一の固定スキーマに従う場合は、このオプションを選択します。
- Avro スキーマ ファイルの Cloud Storage ファイル: トピック内のすべてのメッセージをデコードするために使用される Avro スキーマ ファイルのパス。
Avro バイナリ エンコード
このオプションを Kafka メッセージ形式として選択した場合は、次の追加設定を行います。
- Avro スキーマ ファイルの Cloud Storage ファイル: トピック内のすべてのメッセージをデコードするために使用される Avro スキーマ ファイルのパス。
JSON
このオプションを Kafka メッセージ形式として選択した場合、他の構成は必要ありません。
Kafka オフセットを指定する
個々のワーカーまたはパイプライン全体を再起動するときにメッセージの再処理を回避するには、[Commit offsets to Kafka] オプションを選択します。これにより、パイプラインは中断した場所から処理を再開し、重複処理やデータ不整合の可能性を回避できます。
[Enter Consumer Group ID] フィールドに、このパイプラインのグループの一意の名前を入力します。ほとんどの場合、パイプラインは各メッセージを 1 回読み取り、再起動可能であることが望ましいです。
[デフォルトの Kafka 開始オフセット] フィールドでは、Dataflow パイプラインに 2 つの開始オフセット オプションが用意されています。次のいずれかを選択します。
Earliest: Kafka トピックの先頭からメッセージを処理します。
最新: 使用可能な最新のオフセットからメッセージの処理を開始します。
宛先の構成
これらのオプションは、データ パイプラインが Cloud Storage にデータを書き込む方法を制御します。
[宛先] に、バケットパスを入力し、出力ファイルのファイル名の接頭辞を含めます。ファイル接頭辞の末尾はスラッシュにする必要があります。例:
gs://test-bucket/test-prefix/[ウィンドウ期間] に、Cloud Storage にデータを書き込む時間枠を入力します。データ処理の要件に基づいて、適切な形式（秒の場合は
Ns、分の場合はNm、時間の場合はNh）を選択します。[書き込むファイルの出力ファイル名の接頭辞] には、各出力ファイルに追加する接頭辞を指定して、整理と識別を容易にすることができます。
[出力シャードの最大数] に 0 を設定します。ファイルの書き込み時に生成されるシャードの数を指定できます。この数を増やすとスループットは向上しますが、シャッフル コストが増加するため、コストも増加します。数をゼロに設定すると、サービスが最適な数を選択します。
デッドレター キューを構成する
破損、互換性のないデータ型、スキーマの不一致が原因で、メッセージを処理できないことがあります。
このようなケースに対応するには、テンプレートでデッドレター キューを有効にして、テーブル名を指定します。テンプレートは、標準化されたスキーマを使用してテーブルを作成します。
暗号化を構成する
デフォルトでは、保存データと転送データはすべてGoogle-owned and Google-managed encryption keyによって暗号化されます。顧客管理の暗号鍵（CMEK）がある場合は、独自の鍵を選択できます。CMEK の構成方法の詳細については、メッセージ暗号化を構成するをご覧ください。
ネットワークを構成する
クラスタのネットワークとサブネットワークは、Dataflow テンプレートで指定する必要があります。テンプレートの [省略可能なパラメータ] セクションでは、Dataflow ワーカーのネットワークを定義できます。
Kafka to Cloud Storage Dataflow テンプレートは、デフォルトでプロジェクトのデフォルト ネットワークに Dataflow ワーカーをプロビジョニングします。Managed Service for Apache Kafka クラスタが Dataflow を介して Cloud Storage にデータを送信できるようにするには、Dataflow ワーカーがクラスタのネットワークにアクセスできることを確認します。
Kafka クラスタがプロジェクトのデフォルト ネットワークのサブネットに接続されていない場合は、Kafka クラスタにプロジェクトのデフォルト ネットワークを使用することをおすすめします。
Dataflow パイプラインでネットワークを設定する方法については、以下をご覧ください。
Dataflow ネットワーキングの構成で問題が発生した場合は、Dataflow ネットワーキングのトラブルシューティング ガイドをご覧ください。
オプションの Dataflow パラメータを構成する
オプション パラメータは、構成が Dataflow ワーカーに与える影響を理解している場合にのみ構成してください。設定が正しくないと、パフォーマンスや費用に影響する可能性があります。各オプションの詳細については、省略可能なパラメータをご覧ください。
モニタリング
Kafka to Cloud Storage の Dataflow テンプレートは、コンソール内でログ、指標、エラーを調べることができるモニタリング エクスペリエンスを提供します。このモニタリング ツールスイートは、Dataflow ユーザー インターフェースの一部として利用できます。
[ジョブ指標] タブでは、カスタム ダッシュボードを作成できます。Kafka to Cloud Storage Dataflow テンプレートでは、次の項目をモニタリングするジョブ指標ダッシュボードを設定することをおすすめします。
スループット: 任意の時点で処理されたデータ量。これは、ジョブを介したデータフローをモニタリングし、パフォーマンスの問題を特定するのに役立ちます。
詳細については、Dataflow スループットのモニタリングをご覧ください。
データの鮮度: データ要素のタイムスタンプとパイプラインでイベントが処理される時間との差（秒単位）。これにより、パフォーマンスとデータソースのボトルネックや頻繁な再試行を特定できます。
詳細については、Dataflow データ鮮度モニタリングをご覧ください。
バックログ: 処理を待機しているバイト数。この情報は、自動スケーリングの判断に役立ちます。
Dataflow モニタリングの詳細については、Dataflow モニタリングのドキュメントをご覧ください。
トラブルシューティング
Dataflow パイプラインでパフォーマンスの問題が発生した場合は、Dataflow に用意されている包括的なトラブルシューティング ツールと診断ツールをご利用ください。
一般的な 2 つのシナリオと、それぞれのトラブルシューティング ガイドは次のとおりです。
Dataflow パイプラインのデバッグの概要については、Dataflow パイプラインのトラブルシューティングとデバッグをご覧ください。
既知の制限事項
このテンプレートでは、認証用の認証情報をスキーマ レジストリに渡すことはできません。
Kafka to Cloud Storage Dataflow ジョブを作成するときに、 Google Cloud プロジェクトが Managed Service for Apache Kafka クラスタを含む同じプロジェクトに設定されていることを確認します。