次の表に、Managed Service for Apache Kafka でサポートされている Kafka Connect コネクタのタイプを示します。これらのコネクタを使用すると、Apache Kafka をアプリケーションや他の Google Cloudサービスと統合できます。
| コネクタ | 説明 | ユースケース |
|---|---|---|
| BigQuery シンク | Kafka トピックから BigQuery テーブルにデータをストリーミングします。 | データ ウェアハウジング、分析 |
| Cloud SQL for PostgreSQL ソース | Cloud SQL for PostgreSQL データベースから行レベルの変更を読み取り、Kafka に書き込みます。 | データベースの統合、データベース イベントのモニタリングと対応 |
| Cloud Storage シンク | Kafka トピックから Cloud Storage バケットにデータをストリーミングします。 | データレイクの取り込み、データ アーカイブ |
| 汎用の PostgreSQL ソース | PostgreSQL データベースから行レベルの変更を読み取り、Kafka に書き込みます。 | データベースの統合、データベース イベントのモニタリングと対応 |
| MirrorMaker 2.0 Checkpoint | Kafka クラスタ間でコンシューマー オフセットをコピーします。 | データ レプリケーション、障害復旧、データ移行 |
| MirrorMaker 2.0 Heartbeat | 定期的なハートビート メッセージを生成してトピックに書き込む | クラスタの健全性、ネットワーク接続、クラスタ間のエンドツーエンドのレプリケーション レイテンシをモニタリングする。 |
| MirrorMaker 2.0 Source | Kafka クラスタ間でトピックとデータの複製を行います。 | データ レプリケーション、障害復旧、データ移行 |
| Pub/Sub シンク | Kafka トピックから Pub/Sub トピックにデータをストリーミングします。 | サービス統合、リアルタイム通知 |
| Pub/Sub ソース | Pub/Sub サブスクリプションから Kafka トピックにメッセージをストリーミングします。 | リアルタイムのデータの取り込み、イベント ドリブン アーキテクチャ |
コンバージョンに至ったユーザー
コンバータは、Kafka レコードデータのシリアル化とシリアル化解除を行います。Kafka トピックにある未加工のバイト形式と、Kafka Connect で使用される内部の構造化データ表現との間で変換を行います。
シンクコネクタの場合、コンバータはトピックのワイヤ形式から Kafka Connect の内部データ形式にデータをデシリアル化します。コネクタは、この形式を使用してターゲット システムに書き込みます。
ソースコネクタの場合、コンバータは Kafka Connect の内部データ形式から Kafka トピックの指定されたワイヤ形式にデータをシリアル化します。
コンバータは、コネクタが外部システムと互換性のある形式で Kafka レコードを読み書きできるようにします。
コネクタを構成するときに、次のプロパティを設定します。
キーコンバータ（
key.converter）: Kafka レコードキーのシリアル化とシリアル化解除に使用するコンバータ。値コンバータ（
value.converter）: Kafka レコード値のシリアル化とシリアル化解除に使用するコンバータ。
コンバータを指定しない場合、デフォルトのコンバータ タイプは org.apache.kafka.connect.converters.ByteArrayConverter です。このコンバータは、データを未加工のバイト形式で渡します。
サポートされているコンバータ
Managed Service for Apache Kafka は、次の組み込みコンバータをサポートしています。
| コンバータ | 形式 |
|---|---|
io.confluent.connect.avro.AvroConverter |
Apache Avro |
org.apache.kafka.connect.converters.BooleanConverter |
ブール値 |
org.apache.kafka.connect.converters.ByteArrayConverter |
バイト配列 デフォルトのコンバータ タイプ。2 つのシステム間でメッセージの正確な内容を保持します。 |
org.apache.kafka.connect.converters.DoubleConverter |
Double |
org.apache.kafka.connect.converters.FloatConverter |
浮動小数点数 |
org.apache.kafka.connect.converters.IntegerConverter |
Integer |
org.apache.kafka.connect.json.JsonConverter |
JSON スキーマのない JSON データの場合は、 |
org.apache.kafka.connect.converters.LongConverter |
長い |
org.apache.kafka.connect.converters.ShortConverter |
短いコピー |
org.apache.kafka.connect.storage.StringConverter |
文字列 |
コンバータの選択は、コネクタのタイプと Kafka に保存するデータによって異なります。詳細については、特定のコネクタのドキュメントをご覧ください。
ToDo リスト
コネクタは、並列で動作する 1 つ以上のタスクを作成してデータを転送します。コネクタが作成するタスク数の上限を設定するには、コネクタの 構成プロパティを設定します。tasks.maxコネクタは、この値よりも少ないタスクを作成する場合があります。
tasks.max の値を増やすと、スループットは向上しますが、リソース消費量（CPU とメモリ）も増加します。最適な値は、ワークロードと、Connect クラスタ ワーカーに割り当てられたリソースによって異なります。シンクコネクタの場合、Kafka トピック パーティションの数も並列処理に影響する可能性があります。
タスクの再起動ポリシー
コネクタのタスク再起動ポリシーを設定できます。これにより、障害が発生したときの動作が決まります。コネクタは次のポリシーをサポートしています。
再起動しない。コネクタは、失敗したタスクを再起動しません。このポリシーはデフォルトの動作です。これは、デバッグや、エラー後に手動介入が必要な場合に便利です。
指数バックオフで再開する。コネクタは、遅延（バックオフ期間と呼ばれる）後に失敗したタスクを再起動します。失敗するたびに、遅延は指数関数的に増加します。このポリシーは、ほとんどの本番環境ワークロードに推奨されます。
指数バックオフ ポリシーを使用する場合は、最小バックオフと最大バックオフの値も設定します。最小バックオフは 60 秒より長く、最大バックオフは 7, 200 秒未満にする必要があります。
変換と述語
Managed Service for Apache Kafka は、デフォルトの Kafka Connect の変換と述語をサポートしています。
変換を使用すると、Managed Service for Apache 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ヘッダーキーが存在するメッセージのみが変換によってドロップされます。