このドキュメントでは、Google Cloudの Kafka Connect コネクタの概要について説明します。各コネクタタイプを使用してデータ ストリームを管理、統合するタイミングを確認します。
これらのコネクタは、Kafka Connect フレームワークを使用して Apache Kafka を他のアプリケーションと統合します。Kafka クラスタとアプリケーション間でデータを取り込んで複製します。使用可能なコネクタタイプは次のとおりです。
MirrorMaker 2.0 コネクタ
ソースコネクタ
Checkpoint コネクタ
Heartbeat コネクタ
BigQuery シンクコネクタ
Cloud Storage シンク コネクタ
Pub/Sub ソースコネクタ
Pub/Sub シンクコネクタ
MirrorMaker 2.0 コネクタは、Kafka クラスタ間のデータ レプリケーションと障害復旧用に特別に設計されています。これにより、ある Kafka クラスタから別の Kafka クラスタへのデータのミラーリングが容易になり、高可用性とフォールト トレランスが実現します。
MirrorMaker 2.0 コネクタは、Managed Service for Apache Kafka クラスタと他の Managed Service for Apache Kafka クラスタまたはセルフマネージド Kafka クラスタ間の接続を確立できます。
他のシンクコネクタとソースコネクタは、Kafka とさまざまなGoogle Cloud サービスを統合するために使用されます。これらのコネクタを使用すると、Managed Service for Apache Kafka クラスタと BigQuery、Cloud Storage、Pub/Sub などの Google Cloud サービス間でデータを転送できます。
始める前に
コネクタを調べて作成する前に、次の内容を理解し、前提条件を満たしていることを確認してください。
Kafka Connect と Connect クラスタに関する実用的な知識。コネクタをデプロイする前に、Connect クラスタを作成する必要があります。
シンク コネクタとソース コネクタの場合は、構成するコネクタの種類に応じて、BigQuery テーブル、Cloud Storage バケット、または Pub/Sub トピックとサブスクリプションを理解していること。
コネクタはこれらの形式を使用して構成されるため、YAML または JSON 構成ファイルに精通していること。
MirrorMaker 2.0 を使用する場面
MirrorMaker 2.0 コネクタは、次のようなシナリオで使用します。
データを移行する: Kafka ワークロードを新しい Managed Service for Apache Kafka クラスタに移動します。
障害から復旧する: 障害が発生した場合にビジネスの継続性を確保するために、バックアップ クラスタを作成します。
データを集約する: 分析目的で、複数の Kafka クラスタのデータを中央の Managed Service for Apache Kafka クラスタに統合します。
MirrorMaker 2.0 の主な機能
- トピック、データ、構成、オフセットを含むコンシューマー グループ、ACL など、必要なすべてのコンポーネントを複製します。
- ターゲット クラスタで同じパーティショニング スキーマを維持するため、アプリケーションの移行が簡素化されます。
- 新しいトピックとパーティションを自動的に検出して複製し、手動構成を最小限に抑えます。
- レプリケーション プロセスの健全性とパフォーマンスを追跡できる、エンドツーエンドのレプリケーション レイテンシなどの重要な指標を提供します。
- データ量が多い場合でも信頼性の高い運用を保証し、ワークロードの増加に対応するために水平方向にスケーリングできます。
- オフセット同期、チェックポイント、ハートビートに内部トピックを使用します。これらのトピックには、
offset.syncs.topic.replication.factorなどの構成可能なレプリケーション係数があり、高可用性とフォールト トレランスが確保されます。
MirrorMaker 2.0 Source コネクタを使用する
MirrorMaker 2.0 ソースコネクタは、Kafka クラスタ（ソース）から別の Kafka クラスタ（ターゲット）にトピックとデータを複製します。
| ソース | ターゲット |
|---|---|
| Managed Service for Apache Kafka クラスタ | Managed Service for Apache Kafka クラスタ |
| Managed Service for Apache Kafka クラスタ | 外部またはセルフマネージド Kafka クラスタ |
| 外部またはセルフマネージド Kafka クラスタ | Managed Service for Apache Kafka クラスタ |
MirrorMaker 2.0 ソース コネクタは、次の移行シナリオをサポートしています。
外部またはセルフマネージド Kafka クラスタから Managed Service for Apache Kafka クラスタにデータを複製または移行する
Managed Service for Apache Kafka クラスタから外部またはセルフマネージド Kafka クラスタにデータを複製または移行する。
障害復旧と高可用性の要件を満たすために、複数のリージョン間で Kafka データを複製します。
MirrorMaker 2.0 チェックポイント コネクタを使用する
MirrorMaker 2.0 チェックポイント コネクタの使用は任意です。最後に正常に消費されたメッセージを示すコンシューマー オフセットをコピーします。このプロセスにより、ターゲット クラスタのコンシューマーは、ソース クラスタと同じポイントから処理を再開できます。
このコネクタは、MirrorMaker 2.0 ソースコネクタが機能するために必要ありません。このコネクタは、ソース クラスタからターゲット クラスタへの切り替え時にダウンタイムを最小限に抑えるために ConsumerGroup 状態の同期が必要な場合にのみ必要です。ソースデータのコピーのみが必要な場合、このコネクタは必要ありません。
MirrorMaker 2.0 チェックポイント コネクタは、次のユースケースで使用します。
クラスタ間で一貫したコンシューマーの状態を維持し、シームレスなフェイルオーバーを可能にするための障害復旧。
重要なシナリオで消費者の進捗状況を保持します。
MirrorMaker 2.0 Heartbeat コネクタを使用する
MirrorMaker 2.0 ハートビート コネクタは、ソース Kafka クラスタで定期的なハートビート メッセージを生成するオプションのコンポーネントです。コネクタは、これらのメッセージを専用のトピック（通常は heartbeats という名前）に書き込みます。
heartbeats トピックをターゲット クラスタに複製するように、MirrorMaker 2.0 ソースコネクタを構成できます。ターゲット クラスタでこのレプリケートされたトピックをモニタリングすることで、トピック レプリケーション フローのステータスとパフォーマンスをモニタリングできます。これにより、他のデータが生成または複製されていない場合でも、クラスタ間の接続とデータフローを確認できます。
Heartbeat コネクタを単独でデプロイしても、レプリケーションの健全性は自動的にモニタリングされません。モニタリングに使用するには、heartbeats トピックを複製し、ターゲット クラスタでの存在とタイムリー性を確認するか、これらのハートビートを使用するモニタリング ツールを使用する必要があります。
MirrorMaker 2.0 ソース コネクタが機能するために、Heartbeat コネクタは必要ありません。MirrorMaker 2.0 Heartbeat コネクタは、次のユースケースで使用します。
MirrorMaker 2 レプリケーションの健全性とステータスをモニタリングします。
生成されたハートビートと使用可能な指標を使用して Cloud Monitoring でアラートを構成し、レプリケーションまたはハートビートが停止したときに通知を受け取ります。
シンク コネクタを使用する
シンクコネクタは、Kafka トピックから他のシステムにデータをエクスポートします。
BigQuery シンクコネクタを使用する
BigQuery Sink コネクタは、Kafka トピックから BigQuery テーブルにデータをストリーミングします。
BigQuery Sink コネクタは、次のユースケースで使用します。
データ ウェアハウジング。分析とレポート作成のためにストリーミング データを BigQuery に読み込みます。
リアルタイム ダッシュボードを強化する BigQuery テーブルへのデータの入力。
Cloud Storage シンクコネクタを使用する
Cloud Storage Sink コネクタは、Kafka トピックから Cloud Storage バケットにデータをストリーミングします。
Cloud Storage Sink コネクタは、次のユースケースで使用します。
データレイクの取り込み。Kafka データをデータレイクに保存して、長期的なアーカイブとバッチ処理を行います。
規制要件を満たすためのデータのアーカイブ。
Pub/Sub シンクコネクタを使用する
Pub/Sub Sink コネクタは、Kafka トピックから Pub/Sub トピックにメッセージをストリーミングします。
Pub/Sub シンクコネクタは、次のユースケースで使用します。
サービス統合。Kafka から、Pub/Sub から使用する他の Google Cloudサービスまたはアプリケーションにデータを送信します。
処理されたデータに基づいてリアルタイムの通知やアクションをトリガーする。
ソース コネクタを使用する
ソースコネクタは、他のシステムから Kafka トピックにデータをインポートします。
Pub/Sub ソースコネクタを使用する
Pub/Sub ソースコネクタは、Pub/Sub サブスクリプションから Kafka トピックにメッセージをストリーミングします。
Pub/Sub ソースコネクタは、次のユースケースで使用します。
リアルタイム データ取り込み。クラウド サービスや他のアプリケーションからデータを取得し、Pub/Sub に公開して Kafka に取り込み、ストリーム処理を行います。
イベント ドリブン アーキテクチャ。Pub/Sub に公開されたイベントに基づいて Kafka ベースの処理をトリガーします。
タスクの再起動ポリシー
コネクタのタスク再起動ポリシーを設定できます。このポリシーは、障害が発生したときの動作を決定します。コネクタは次のポリシーをサポートしています。
再起動しない。コネクタは失敗したタスクを再起動しません。このポリシーはデフォルトの動作です。これは、デバッグや、エラー後に手動での介入が必要な状況で役立ちます。
指数バックオフで再起動します。コネクタは、遅延（バックオフ期間）後に失敗したタスクを再起動します。遅延は、失敗するたびに指数関数的に増加します。このポリシーは、ほとんどの本番環境ワークロードに推奨されます。
指数バックオフ ポリシーを使用する場合は、最小バックオフと最大バックオフの値も設定します。最小バックオフは 60 秒より大きく、最大バックオフは 7, 200 秒より小さくする必要があります。
変換と述語
Kafka Connect は、デフォルトの Kafka 変換と述語をサポートしています。
構成は、コネクタ構成の一部として指定します。たとえば、DoNotProcess ヘッダーキーを含むメッセージを無視するように シンク コネクタを構成するには、次の構成をコネクタに追加します。
transforms=dropMessage
transforms.dropMessage.type=org.apache.kafka.connect.transforms.Filter
transforms.dropMessage.predicate=hasKey
predicates=hasKey
predicates.hasKey.type=org.apache.kafka.connect.transforms.predicates.HasHeaderKey
predicates.hasKey.name=DoNotProcess
この構成では、次の処理が行われます。
org.apache.kafka.connect.transforms.predicates.HasHeaderKey型のhasKeyという名前の述語を構成します。この述語は、キーDoNotProcessを含むヘッダーを含むすべてのメッセージに一致します。org.apache.kafka.connect.transforms.Filter型のdropMessageという名前の変換を構成します。この変換では、構成された述語に一致するすべてのメッセージが削除されます。変換を述語
hasKeyにリンクします。これにより、DoNotProcessヘッダーキーが存在するメッセージのみが変換によってドロップされます。
詳細については、変換と述語に関する Kafka のドキュメントをご覧ください。
次のステップ
MirrorMaker 2.0 コネクタを作成する
BigQuery シンクコネクタを作成する
Pub/Sub ソースコネクタを作成する
Pub/Sub シンクコネクタを作成する