LASSIC Media らしくメディア

LASSIC IT事業部｜元請（プライムベンダー）としてシステム保守・運用を受託

Kafka Connectとは何か

Kafka Connectとは、Apache Kafka^*1とデータベース・検索インデックス・ファイルシステムなどの外部システムとの間でデータを連携させるための、Apache Kafka本体に含まれるオープンソースのフレームワークを指します^*1。コネクタと呼ぶ設定単位でシステム間のデータコピーを定義し、実際の処理はWorkerと呼ばれるプロセス群が担います。

この記事では、Kafka Connect単体の仕組み（コネクタ・Worker・SMT・オフセット管理）に絞って解説します。コスト最適化の観点でのマネージドストリーミング基盤の比較や、変更データキャプチャ（CDC）ツール単体の解説は本稿の対象外です。データベースの変更を検知する仕組みそのものに関心がある場合は、CDCの専門解説を参照するのが近道です。

Source・Sinkコネクタが担うデータの入口と出口

最初に、Kafka ConnectでKafkaとの連携方式を検討する際に押さえるべき直接の回答を示します。Kafka Connectは、Sourceコネクタで外部システムのデータをKafkaトピックへ取り込み、SinkコネクタでKafkaトピックのデータを外部システムへ書き出す、2種類のコネクタでデータフローを構成する仕組みです^*2。

Sourceコネクタは、データベースやメッセージキューなど連携元のシステムからデータを継続的に取り込み、Kafkaのトピックへストリームとして書き込む役割を持ちます^*2。代表例として、リレーショナルデータベース全体をKafkaに取り込み、テーブルの更新を継続的にストリーミングする用途が挙げられます^*2。

一方のSinkコネクタは、Kafkaトピックのデータを検索インデックスやバッチ処理システムなど連携先へ配信する役割です^*2。検索エンジンへのデータ配信や、分散ファイルシステムへのバッチ出力といった用途で使われます^*2。

コネクタ自体はデータのコピー処理を直接実行しません。コネクタの設定はコピー対象のデータを定義するだけであり、実際の処理はコネクタが分割した「タスク」という単位に委ねられます^*3。タスクを複数に分割することで、並列処理とスケーラビリティを実現しているのがKafka Connectの設計思想です^*3。

コネクタとプラグインの区別

Kafka Connectの文脈では「コネクタプラグイン」と「コネクタインスタンス」を区別する必要があります^*3。プラグインは連携先ごとに用意されるコード一式であり、インスタンスはそのプラグインに具体的な接続先・認証情報・トピック名などを設定して起動した実行単位です。同じプラグインから複数のインスタンスを生成し、異なるテーブルや異なる連携先に向けて並行運用することもできます。

Worker — standalone/distributedモードの違い

コネクタとタスクを実際に実行するプロセスがWorkerです。Kafka Connectには、standalone（単一プロセス）モードとdistributed（分散）モードの2つの実行形態があります^*4。

standaloneモードは、単一のプロセスがすべてのコネクタとタスクを実行する形態です^*3。起動時にはWorker設定ファイルとコネクタ設定ファイルを引数として渡すコマンド構成になっており、複数のコネクタ設定を同時に指定して起動することもできます^*5。開発・検証用途に向いた構成で、スケーラビリティと障害耐性は限定的です^*3。オフセット（後述）はローカルファイルシステム上のファイルに保存されるため、複数Workerで同じファイルを共有すると、オフセットデータが上書き・消失するリスクがある点に注意が必要です^*5。

distributedモードは、複数のWorkerプロセスが自動的にコネクタとタスクの負荷分散を担う形態です^*4。Workerを追加・停止した場合や、Workerが予期せず停止した場合は、残りのWorkerがその状況を認識し、協調して作業を再配分します^*3。設定・オフセット・タスクステータスの両面で自動的なフォールトトレランスを提供するのが、distributedモードの利点です^*4。

distributedモードのWorkerは、同一のgroup.idを持つことで互いを発見し、1つのKafka Connectクラスタを構成します^*5。起動コマンドはWorker設定ファイルのみを読み込む形式で、standaloneのようにコネクタ設定を個別に指定する必要はありません^*5。コネクタの追加・削除・設定変更は、REST API経由で動的に行う設計になっています。

タスクの再分配とリバランス

コネクタを最初にクラスタへ投入した時点や、Workerの増減が発生した時点で、Kafka ConnectはWorker間のリバランスを実行します^*3。近年のバージョンでは「incremental cooperative rebalancing」という方式が採られており、新規に必要となったタスク・削除されたタスク・移動が必要なタスクだけが影響を受け、無関係なタスクの処理は継続されます^*6。Workerがクラスタから外れてもすぐに再分配せず、一定時間（デフォルトで数分程度の遅延)復帰を待つ設定も備えています^*6。個々のタスクが失敗した場合は自動再起動されないため、REST API経由で手動再起動する運用が前提になっています^*3。

Converter・SMTでデータ形式と内容を制御

Kafka Connectとは、タスクがバイト列とConnect内部のデータ形式との間で変換する仕組みも含んだフレームワークです。この変換を担うのがConverterです^*3。タスクはConverterを介して、外部システムのデータ形式をバイト列からConnect内部形式に、またその逆方向に変換します^*3。Avro・Protobuf・JSON Schema・素のJSON・文字列・バイト配列など、複数の形式に対応したConverterが提供されています^*3。

Converterの選択は、SourceコネクタとSinkコネクタの間でスキーマの互換性を保つ上で重要な設計判断です。スキーマレジストリと組み合わせたAvroやProtobufのConverterを使うと、スキーマの進化を無理なく管理しやすくなります。JSON Converterはスキーマレジストリを必要としない代わりに、スキーマ変更時の互換性検証は自前で担う必要があります。

SMT（Single Message Transforms）の役割

SMT（Single Message Transforms）は、個々のメッセージに対して簡易な変換を適用する仕組みです^*7。SMTは1件のレコードを入力として受け取り、変更後のレコードを出力する単純な関数として動作します^*3。フィールドのマスキングや不要なフィールドの削除、トピック名のルーティング変更といった軽量な加工を、別のストリーム処理アプリケーションを用意せずに実現できます^*7。

SMTはSourceコネクタがレコードを生成した後、Kafkaに書き込まれる前の段階で適用されます^*7。Sinkコネクタの場合は逆に、Kafkaから読み込んだメッセージがコネクタへ渡される前に適用される仕組みです^*7。複数のSMTを設定した場合はチェーン状に連結され、最初のSMTの出力が次のSMTの入力になる形で順次処理されます^*7。標準で提供されるSMTで要件を満たせない場合は、自作のSMTを実装して組み込むことも可能です^*7。

オフセット管理とスケーリング・障害耐性

オフセットとは、コネクタがどこまでデータを処理したかを示す位置情報です。distributedモードでは、コネクタ設定・オフセット・タスクステータスの3種類の情報を、それぞれ専用のKafkaトピックに保存します^*5。具体的には、コネクタ設定を保存するconfig.storage.topic、オフセット情報を保存するoffset.storage.topic、タスクの状態を保存するstatus.storage.topicの3トピックです^*5。同一のgroup.idを持つWorker群は、これら3つの内部トピックを共有します^*5。

standaloneモードでは事情が異なります。オフセットはKafkaのトピックではなく、ローカルファイルシステム上のファイル(offset.storage.file.filenameで指定)に保存されます^*5。この方式はプロセス再起動時に前回の続きから処理を再開できる一方、単一プロセス・単一ホストに依存するため、ホスト障害がそのまま処理停止につながります。

スケーリングの観点では、distributedモードのWorkerを追加するだけでクラスタ全体の処理能力を増強できます^*4。追加されたWorkerには既存タスクの一部が自動的に再配分され、コネクタ設定を個別に変更する必要はありません^*3。障害耐性についても、Workerが1台停止しても残りのWorkerがタスクを引き継ぐため、単一障害点を避けられる設計です^*4。

Dead Letter Queueによるエラー処理

Sinkコネクタの処理でレコードの変換やConverterの処理に失敗した場合の挙動も、設定で制御できます。デフォルトのerrors.tolerance=noneでは、エラー発生時にコネクタ全体を即座に失敗させます^*3。errors.tolerance=allに変更すると、エラーとなったレコードを許容してタスクを継続させることができます^*3。あわせてerrors.deadletterqueue.topic.nameでDead Letter Queue用のトピックを指定すれば、処理に失敗したレコードを退避先トピックへ振り分け、後から原因を調査する運用が可能になります^*3。

CDC・DWH連携でのKafka Connectの位置づけ

Kafka Connectは、CDC（Change Data Capture、データベースの変更をリアルタイムに検知する仕組み）を実現する専用ツール群の実行基盤としても使われます。データベースの変更ログを検知してKafkaに流し込むタイプのSourceコネクタは、Kafka Connectのプラグインとして動作する構成が一般的です。CDCの検知ロジック自体の詳細は別テーマとなるため、本稿ではKafka Connectがそれらのコネクタを実行・管理する基盤である点に留めます。

DWH（データウェアハウス）への連携では、Kafkaに集約されたデータをSinkコネクタで各DWHに書き出す構成が典型的です。Sourceコネクタで複数の業務システムからデータを集約し、SMTで不要なフィールドを除去したうえでSinkコネクタへ渡すという流れを組むことで、業務システムとDWHを疎結合に保ったまま連携を仕組み化できます。この位置づけにおいて、Kafka ConnectはETL/ELTツールが担う変換処理そのものではなく、システム間のデータ移動を標準化された設定で担う「配管」の役割を持つと理解すると整理しやすくなります。

マネージドとセルフホスト、どちらを選ぶか

Kafka Connectの実行環境は、大きくセルフホスト型とマネージド型に分かれます。セルフホストのKafka Connectは、開発・検証用のstandalone環境から始め、大規模なデータパイプラインを支える本番環境まで段階的にスケールアップできる構成です^*8。低い導入コストで始められる一方、Worker・トピック・監視の運用は自社の責任範囲になります^*8。

Confluent Cloudのようなマネージド型サービスでは、クラウドオブジェクトストレージ・リレーショナルDB・NoSQL・CDCソース・DWH・メッセージキューなど多様な連携先向けに、事前構築済みのフルマネージドコネクタが提供されています^*9。コンソールやCLIから設定するだけで弾力的にスケールし、Workerクラスタそのものの運用負荷を負わない点が特徴です^*9。フルマネージドのコネクタ一覧にまだ対応先が存在しない連携が必要な場合は、セルフホストのKafka Connectクラスタで自前運用する選択肢が案内されています^*9。

外注と内製の判断軸

Kafka Connectの導入を内製で完結させる場合、必要になる知識は幅広いです。Kafka本体のトピック設計・パーティション設計、Connect Workerのクラスタ構成、Converterとスキーマ管理、SMTやカスタムコネクタの実装、そして障害時のリバランス挙動の理解まで、複数領域の専門知識を持つ人材を確保する必要があります。

設計や運用設計を誤ると、想定外の影響が生じます。オフセットの保存先設計を誤ると、Worker障害時にデータの重複処理や欠落が発生し、下流のDWHやCDC連携先でデータ不整合が起きる要因になります。SMTの変換ロジックに不備があると、後段のSinkコネクタでスキーマ不一致によるタスク失敗が連鎖することもあります。こうした障害は本番運用が始まってから顕在化しやすく、事前の設計レビューが欠かせません。

専門パートナーに依頼する場合と内製する場合の違いは、設計段階での経験値に表れやすい部分です。パートナーに依頼すれば、コネクタ選定・Worker構成・SMTの設計パターンを踏まえた構成レビューを受けられます。内製の場合は、これらの設計判断を自社で積み上げる時間とリスクを引き受ける形になります。どちらを選ぶかは、社内にKafka Connectの運用経験を持つ人材がいるか、他のシステム開発と並行して習熟の時間を確保できるかで判断するのが現実的です。

LASSICでは、Kafka Connectを含むデータ連携基盤の設計・構築・保守運用を、元請（プライムベンダー）としてお客様のシステム保守・運用案件の中で担っています^*10。コネクタ選定からWorkerのクラスタ構成、SMTを用いたデータ加工の設計まで、要件に応じた体制で支援します。

まとめ：Kafka Connect導入の3つの判断軸

本稿では、Kafka ConnectのSource/Sinkコネクタ・Worker・Converter・SMT・オフセット管理の仕組みを整理しました。要点を3つに集約すると、第一にコネクタとタスクの分離によって並列処理とスケーラビリティを確保する設計であること、第二にdistributedモードのオフセット管理とリバランスが障害耐性の中核を担うこと、第三にマネージド型とセルフホスト型のどちらを選ぶかは運用体制の有無で判断すべきことです。これらを踏まえたうえで、内製と外注のどちらが自社に適するかを検討することが、導入後の運用負荷を左右します。

よくある質問

著者：テレリモ総研編集部　鈴木 亮佑

ご不明な点はお問い合わせフォームからもご連絡いただけます。

1. *1　出典：Apache Software Foundation「Apache Kafka Documentation – Kafka Connect」（https://kafka.apache.org/documentation/#connect）
2. *2　出典：Confluent「Kafka Connect Concepts for Confluent Platform」（https://docs.confluent.io/platform/current/connect/concepts.html）
3. *3　出典：Confluent「Kafka Connect Concepts for Confluent Platform」（https://docs.confluent.io/platform/current/connect/concepts.html）
4. *4　出典：Apache Software Foundation「Kafka Connect Administration」（https://kafka.apache.org/42/kafka-connect/administration/）
5. *5　出典：Confluent「Configure Self-Managed Kafka Connect for Confluent Platform – User Guide」（https://docs.confluent.io/platform/current/connect/userguide.html）
6. *6　出典：Apache Software Foundation「Kafka Connect Administration」（https://kafka.apache.org/42/kafka-connect/administration/）
7. *7　出典：Confluent「Get started with Single Message Transformations for self-managed connectors」（https://docs.confluent.io/kafka-connectors/transforms/current/overview.html）
8. *8　出典：Confluent「Kafka Connectors for Confluent Platform」（https://docs.confluent.io/platform/current/connect/kafka_connectors.html）
9. *9　出典：Confluent「Confluent Cloud Fully-Managed Connectors Overview」（https://docs.confluent.io/cloud/current/connectors/overview.html）
10. *10　出典：LASSIC株式会社（自社提供体制情報）

Tweet