オープンソースの Apache Kafka トピックから Managed Service for Apache Kafka にデータを移動する必要がある場合は、Dataflow テンプレートを使用して移動できます。 Google Cloud コンソール、REST API、または Google Cloud CLI を使用できます。Kafka トピックのメッセージは、AVRO または JSON スキーマとエンコードを使用して公開する必要があります。
このドキュメントでは、 Google Cloud コンソールを使用して Kafka to Kafka Dataflow テンプレートを構成する方法について説明します。
Google Cloud 個の商品を使用
Kafka to Kafka Dataflow テンプレートは、次の課金対象の Google Cloud プロダクトを使用します。料金計算ツールを使用すると、予想使用量に基づいて費用の見積もりを出すことができます。
- Dataflow: Dataflow はフルマネージドのデータ処理サービスです。Kafka to Kafka Dataflow テンプレートは、Dataflow を使用して、外部 Kafka トピックからデータを読み取り、Managed Service for Apache Kafka トピックに書き込むパイプラインを作成します。Dataflow の自動スケーリング機能と自己修復機能により、パイプラインが信頼性と効率性を維持して実行されます。
- Cloud Storage: Cloud Storage はオブジェクト ストレージを提供します。外部 Kafka ソースに TLS 認証を使用する場合にのみ使用されます。TLS 証明書を含むキーストア ファイルとトラストストア ファイルを Cloud Storage に保存し、Dataflow パイプラインが Cloud Storage からアクセスします。
- Secret Manager: Secret Manager は、シークレットと認証情報の管理サービスです。このプロダクトは、外部 Kafka ソースに TLS 認証または SASL_PLAIN 認証を選択した場合にのみ使用されます。ユーザー名、パスワード、証明書などの認証情報を保存します。これらの認証情報は、外部 Kafka クラスタに安全に接続するために使用されます。
また、このソリューションでは Managed Service for Apache Kafka も使用します。
- Managed Service for Apache Kafka: Apache Kafka の実行に役立つ Google Cloud サービス。このソリューションでは、Managed Service for Apache Kafka がデータの宛先として使用されます。外部の Kafka トピックから Managed Service for Apache Kafka クラスタ内のトピックにデータをストリーミングします。Managed Service for Apache Kafka の料金の詳細については、料金ガイドをご覧ください。
始める前に
Kafka to Kafka Dataflow テンプレートを起動する前に、次の操作を完了していることを確認してください。
Managed Service for Apache Kafka クラスタとトピックを作成します。
クラスタとトピックを作成する方法の 1 つは、Managed Service for Apache Kafka クイックスタートに沿って操作することです。
トピックに Avro レコードが含まれている場合は、追加のリソース要件について、Kafka メッセージ形式を構成するをご覧ください。
次の Google Cloud API を有効にします。
Dataflow
Secret Manager（外部 Kafka ソースに SASL_PLAIN または TLS 認証を使用する場合のみ）
Cloud Storage（外部 Kafka ソースに TLS 認証を使用する場合のみ）
gcloud services enable dataflow.googleapis.com \ secretmanager.googleapis.com storage.googleapis.com
SASL_PLAIN 認証リソースを準備する
外部 Kafka ソースに SASL_PLAIN 認証情報を使用する場合は、このセクションの手順に沿って操作します。
Google Cloud コンソールで、Secret Manager に移動します。
Kafka ユーザー名用のシークレットと Kafka パスワード用のシークレットの 2 つを作成します。
シークレットごとに、「kafka-username」や「kafka-password」などのわかりやすい名前を指定します。
必要なロールと権限など、シークレットの作成方法の詳細については、シークレットを作成するをご覧ください。
TLS 認証リソースを準備する
外部 Kafka ソースに TLS 認証情報を使用する場合は、このセクションの手順に沿って操作します。
Google Cloud コンソールで、Secret Manager に移動します。
キーストア ファイル（JKS 形式）に、Kafka クラスタでクライアントを認証するために必要な TLS 証明書と秘密鍵が含まれていることを確認します。
トラストストア ファイル（JKS 形式）に、Kafka ブローカーの ID 検証に必要な信頼された証明書が含まれていることを確認します。
キーストア ファイルとトラストストア ファイルを Cloud Storage にアップロードします。
Cloud Storage にファイルをアップロードする方法については、ファイル システムからオブジェクトをアップロードするをご覧ください。
TLS 構成用に 3 つのシークレットを作成します。
トラストストアのパスワード: トラストストア ファイルへのアクセスに使用されるパスワードを保存します。
キーストア パスワード: キーストア ファイルへのアクセスに使用するパスワードを保存します。
秘密鍵のパスワード: キーストア ファイル内の秘密鍵へのアクセスに使用されるパスワードを保存します。
各シークレットにわかりやすい名前を付けます。
必要なロールと権限など、シークレットの作成方法の詳細については、シークレットを作成するをご覧ください。
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 Kafka Dataflow テンプレートを起動する
コンソールのクラスタの詳細ページから Kafka to Kafka Dataflow テンプレートを起動できます。
-
Google Cloud コンソールで、[クラスタ] ページに移動します。
プロジェクトで作成したクラスタが一覧表示されます。
- クラスタの詳細ページを表示するには、クラスタ名をクリックします。
- クラスタの詳細ページで、[データをインポート] をクリックします。
[テンプレート「Kafka to Kafka」を使用して Dataflow ジョブを作成する] ページが開きます。
以降のセクションの情報に従って、テンプレートのフィールドを構成します。
ジョブ名を入力する
[ジョブ名] フィールドに、Dataflow ジョブの名前を入力します。
名前は、プロジェクトで現在実行中のすべてのジョブ間で一意である必要があります。
パイプラインのリージョン エンドポイントを選択する
[リージョン エンドポイント] フィールドで、リージョン エンドポイントを Kafka クラスタのロケーションに設定して、リージョン間のデータ転送費用を最小限に抑えます。
Dataflow ワーカーは、Kafka クラスタのリージョンとは関係なく実行できます。ただし、Kafka クラスタのリージョン外でワーカーを起動すると、リージョン間の下り（外向き）料金が発生します。
クラスタのロケーションを表示するには、Managed Service for Apache Kafka クラスタを一覧表示するの手順に沿って操作します。
ソースの構成
[ソース] で [セルフマネージドまたは外部 Kafka] を選択します。
[Kafka ブートストラップ サーバー] に、外部 Kafka クラスタのブートストラップ サーバーのアドレスを入力します。
[Source Kafka Bootstrap server and topic] に、外部 Kafka ブートストラップ サーバーとトピックのアドレスを入力します。
[Kafka ソース認証モード] で、表示された選択肢から認証モードを選択します。外部 Kafka の場合、次の 3 つのオプションがあります。
SASL_PLAIN
TLS
なし
SASL_PLAIN Kafka ソース認証の場合
このセクションに進む前に、SASL_PLAIN 認証リソースを準備するをご覧ください。
Kafka ソース認証として SASL_PLAIN を選択した場合は、次の追加フィールドに入力します。
[Kafka SASL/PLAIN ユーザー名のシークレット バージョン ID] に、ユーザー名のバージョン ID を入力します。
[Secret version ID for Kafka SASL/PLAIN password] に、パスワードのバージョン ID を入力します。
TLS Kafka ソース認証の場合
このセクションに進む前に、TLS 認証リソースを準備するをご覧ください。
Kafka ソース認証として TLS を選択した場合は、次の追加フィールドに入力します。
[キーストアの場所] に、Java KeyStore（JKS）ファイルの Cloud Storage パスを入力します。このファイルには、Kafka クラスタでパイプラインを認証するために必要な TLS 証明書と秘密鍵が含まれています。例: gs://your-bucket/keystore.jks
[トラストストア ファイルの場所] フィールドに、Java Truststore（JKS）ファイルへの Cloud Storage パスを入力します。このファイルには、Kafka ブローカーの ID 検証に使用される信頼された証明書が含まれています。
[トラストストア パスワードのシークレット バージョン ID] に、トラストストア JKS ファイルにアクセスするためのパスワードを保存する Secret Manager シークレット ID を指定します。
[キーストア パスワードのシークレット バージョン ID] に、JKS ファイルにアクセスするためのパスワードを保存する Secret Manager シークレット ID を指定します。
[秘密鍵パスワードのシークレット バージョン ID] フィールドに、キーストア内の秘密鍵に個別のパスワードがある場合は、そのパスワードを含む Secret Manager のシークレット ID を入力します。
Kafka にオフセットを commit する
この設定は、外部 Kafka ソースのどのメッセージが処理されたかをパイプラインが追跡する方法を制御します。この設定を有効にすると、パイプラインでメッセージが確実に処理され、再起動時のエラーを回避できます。
Kafka はオフセットを使用してメッセージを追跡します。これらは、各メッセージ ストリーム内のパイプラインの位置を記憶するブックマークと考えることができます。この設定を有効にすると、パイプラインはブックマークを Kafka に定期的に保存します。エラー、更新、定期メンテナンスが原因でパイプラインが再起動した場合、保存されたブックマークを使用して、中断したところから再開できます。これにより、メッセージの欠落や同じメッセージの誤った 2 回処理を防ぐことができます。
[Kafka へのオフセット commit] を有効にします。
コンシューマー グループ ID を指定します。これは、Kafka にパイプラインを識別する名前です。
[デフォルトの Kafka 開始オフセット] で、次のいずれかを選択します。
Earliest: Kafka トピックの先頭からメッセージを処理します。
最新: 最新のメッセージからメッセージの処理を開始します。
宛先の構成
宛先として [Managed Service for Apache Kafka] を選択します。
[Kafka クラスタ] で、作成済みのクラスタを選択します。
既存のトピックを選択することも、新しいトピックを作成することもできます。
[Kafka destination autheticaton method] で、[アプリケーションのデフォルト認証情報] を選択します。
暗号化を構成する
デフォルトでは、すべての保存データと転送データはGoogle-owned and Google-managed encryption keyによって暗号化されます。顧客管理の暗号鍵（CMEK）がある場合は、独自の鍵を選択できます。CMEK の構成方法の詳細については、メッセージ暗号化を構成するをご覧ください。
ネットワークを構成する
Dataflow テンプレートで、クラスタのネットワークとサブネットワークを指定する必要があります。テンプレートの [省略可能なパラメータ] セクションでは、Dataflow ワーカーのネットワークを定義できます。
Kafka to Kafka Dataflow テンプレートは、デフォルトでプロジェクトのデフォルト ネットワークに Dataflow ワーカーをプロビジョニングします。外部 Kafka クラスタが Dataflow を介して Managed Service for Apache Kafka にデータを送信できるようにするには、Dataflow ワーカーがクラスタのネットワークにアクセスできることを確認します。
Kafka クラスタがプロジェクトのデフォルト ネットワークのサブネットに接続されていない場合は、Kafka クラスタにプロジェクトのデフォルト ネットワークを使用することをおすすめします。
Dataflow パイプラインでネットワークを設定する方法については、以下をご覧ください。
Dataflow ネットワーキングの構成で問題が発生した場合は、Dataflow ネットワーキングのトラブルシューティング ガイドをご覧ください。
オプションの Dataflow パラメータを構成する
Dataflow ワーカーに対する構成の影響を把握している場合にのみ、省略可能な Dataflow パラメータを構成します。設定が正しくないと、パフォーマンスや費用に影響する可能性があります。各オプションの詳細については、オプションのパラメータをご覧ください。
モニタリング
Kafka から Managed Service for Apache Kafka への Dataflow テンプレートは、コンソール内でログ、指標、エラーを調べることができるモニタリング エクスペリエンスを提供します。このモニタリング ツールスイートは、Dataflow ユーザー インターフェースの一部として利用できます。
[ジョブ指標] タブでは、カスタム ダッシュボードを作成できます。Kafka to Managed Service for Apache Kafka テンプレートの場合は、次の項目をモニタリングするジョブ指標ダッシュボードを設定することをおすすめします。
スループット: 任意の時点で処理されたデータ量。これは、ジョブのデータフローをモニタリングし、パフォーマンスの問題を特定するのに役立ちます。
詳細については、Dataflow スループットのモニタリングをご覧ください。
データの更新速度: データ要素のタイムスタンプとパイプラインでイベントが処理される時間との差（秒単位）。これにより、パフォーマンスとデータソースのボトルネックや頻繁な再試行を特定できます。
詳細については、Dataflow の鮮度モニタリングをご覧ください。
バックログ: 処理待ちのバイト数。この情報は、自動スケーリングの決定に役立ちます。
Dataflow モニタリングの詳細については、Dataflow モニタリングのドキュメントをご覧ください。
トラブルシューティング
Dataflow パイプラインでパフォーマンスの問題が発生した場合は、Dataflow に用意されている包括的なトラブルシューティング ツールと診断ツールを使用できます。
一般的なシナリオと、それぞれのトラブルシューティング ガイドは次のとおりです。
Dataflow パイプラインのデバッグの概要については、Dataflow パイプラインのトラブルシューティングとデバッグをご覧ください。