CData Sync は、SQL Server や Salesforce をはじめとする 400 種類以上のデータソースからのデータ連携に対応しています。その同期先の一つとして、従来から Apache Kafka を利用できました。ただし、メッセージのシリアライズに Schema Registry の指定が必須という制約があり、これが Kafka への連携を試す際のハードルになっていました。

CData Sync V26.3 では、この制約が解消され、Schema Registry なしでも Kafka への連携を開始できるようになります。あわせて、トピックへのルーティング方式も強化されました。多様なデータソースを持つ環境でも、これらのデータをまとめて Kafka へ連携しやすくなっています。

本記事の検証には CData Sync 26.3.9736.0 を使用しています。

CData Sync V26.3 での Kafka 同期先強化のポイント

CData Sync は SQL Server や Salesforce、Oracle など 400 種類以上のデータソースに対応しています。それらのデータを共通の設定手順で Kafka へ連携できます。CData Sync V26.3 における Kafka 同期先の強化は、大きく次の 2 点です。

• Schema Registry の任意化

• トピックルーティングの 3 モードからの選択（単一トピック、テーブルごとトピック、テンプレートトピック名）

Schema Registry を任意に

Kafka でやり取りするメッセージには、あらかじめ構造（スキーマ）を決めておきたい場合があります。そのための仕組みが Schema Registry です。Schema Registry を使うと、発行するメッセージが Avro など決められた形式に沿っていることを保証できます。その分、Schema Registry 自体を別途構築・運用する必要があり、これが導入のハードルになっていました。

これまで Kafka 同期先接続では、メッセージの形式を判定する方法として Schema Registry の指定が必須で、それ以外の設定では接続エラーになっていました。CData Sync V26.3 ではこの制約が解消されています。JSON など Schema Registry を使わないメッセージフォーマットを選択すれば、Schema Registry の URL を設定しなくても接続・同期できるようになりました。

これにより、Schema Registry を構築していない Kafka クラスタに対しても、接続テストからジョブ実行までを行えるようになりました。発行される JSON メッセージは、行データと操作メタデータ（insert・update・delete のいずれか）を含む有効な JSON オブジェクトとして構成されます。主キーを持たないテーブルであっても、メッセージのキー自体は不要なため、そのまま JSON メッセージとして発行できます。

トピックルーティングを柔軟に選択

CData Sync V26.3 では、ソーステーブルから Kafka トピックへのマッピング方式を 2 モードから選択できるようになりました。

指定したトピックが存在しない場合は、ブローカー側の設定に応じて自動作成されるか、エラーメッセージが返されます。トピックルーティングの設定は、JSON・Avro といったメッセージフォーマットの選択とは別に行えます。

補足として、テーブルごとトピックの命名規則には Avro 互換性の観点から改善が加えられています。Avro のレコード名は英数字とアンダースコアのみが許容される仕様です（正規表現 ^[A-Za-z_][A-Za-z0-9_]*$）。そのため、従来のスキーマ名とテーブル名をハイフンで連結する命名（例 dbo-Employee）は、無効な文字列になってしまう問題がありました。V26.3 ではアンダースコアでの連結に変更されています（例 dbo_Employee）。

実際の設定手順

CData Sync は SQL Server、Oracle、Salesforce など 400 種類以上のデータソースに対応しています。ここでは、連携したいデータソースへの接続はすでに作成済みであることを前提に、Kafka 同期先まわりの設定手順を中心にご紹介します。データソース側の接続作成については、各コネクタのドキュメントをご参照ください。

Step 1　同期先（Apache Kafka）の接続設定

画面左側の「接続」タブから「接続を追加」を選択し、Kafka を選択して任意の接続名を入力します。主な接続プロパティとして以下の項目を入力します。

各項目を入力後、画面右上の「保存およびテスト」で接続が正常に確立できることを確認します。

Step 2　ジョブの作成とトピックルーティングの設定

「Jobs」タブから「Add Job」を選択し、任意のジョブ名を入力したうえで、データソースの接続と Step 1 で作成した Kafka の接続を選択します。ジョブ作成時に、トピックルーティングモードを選択します。今回は「ソーステーブルごとに 1 つのトピック」を選択します。

Step 3　ジョブの実行と発行メッセージの確認

ジョブ画面上部の「Run」ボタンをクリックしてジョブを実行します。実行後、「History」タブから各テーブルの転送件数や実行時間を確認できます。

発行された JSON メッセージの内容は、Kafka に付属する kafka-console-consumer コマンドで確認できます。ブラウザベースのビューアーツールを使えば、GUI 上で確認することもできます。

kafka-console-consumer --bootstrap-server localhost:9092 --topic orders --from-beginning

まとめ

CData Sync V26.3 の Kafka 同期先強化により、Schema Registry を用意していない環境でも Kafka への連携を手軽に開始できるようになりました。トピックルーティングを柔軟に選べるため、コンシューマ側の要件やシステム構成に合わせて連携方式を調整できます。ぜひ CData Sync V26.3 をお試しください。

CData Sync の 30 日間無償トライアルはこちらからダウンロードいただけます。

また、使っていて気になった点やご不明な点があれば、お気軽にテクニカルサポートまでお問い合わせください。