Apacheコンポーネントを活用したCDCソリューション

1. はじめに

昨今のビジネスシーンでは、リアルタイム性の高いデータ活用や、分散環境でのデータ同期・分析が求められています。
オンラインショッピングやIoTデバイス、金融取引など、取り扱うデータ量が増加する一方で、可用性やスピードを維持するためには、従来のバッチ処理だけでは不十分なケースも多くなってきました。
そのような環境下で注目されているのが、CDC（Change Data Capture） という技術です。

本記事では、CDCの基本的な概念、どのような背景から注目されているのか、そしてCDCソリューションとしてオープンソースのApacheコンポーネントを活用するメリットについて解説していきます。

1.1 CDC（Change Data Capture）とは

CDC（Change Data Capture） とは、データベースなどのデータソースにおける変更（Create、Update、Delete）を継続的に検知し、その変更履歴を別のシステムへ伝搬するための技術です。
従来は一定時間ごとにバッチ処理でソースとターゲットのデータを同期する手法が中心でしたが、近年の要求としては、「ほぼリアルタイムで変更を把握したい」「大規模なデータを効率よく処理したい」といったニーズが高まっています。そのため、トランザクションログから変更分だけを抽出し、ストリーミングとして転送・分析を行うCDCが注目を集めています。

CDC のメリットは、以下のような点が挙げられます。

• リアルタイム性: 変更があったタイミングでイベントを生成するため、常に最新情報を反映可能
• 軽量・効率的: 差分だけを扱うため、フルデータのコピーに比べ通信量や処理負荷が少ない
• データ一貫性の維持: トランザクションログ単位での取得により、整合性の確保が容易

1.2 Apacheコンポーネントを活用するメリット

CDCを実装する場合、商用のデータ統合ツールを用いる方法もありますが、近年はオープンソースのApacheプロジェクトを中心としたソリューションが非常に注目を集めています。

主なメリットとして、以下の点が挙げられます。

1. 高い拡張性とコミュニティサポート
   Apache Kafka や Apache NiFi、Apache Flink など、エコシステムが非常に豊富で大規模コミュニティを持つプロジェクトが多数存在します。多種多様なコネクタやプラグイン、拡張モジュールが公開されており、自社の環境に合わせて組み合わせる柔軟性が高いです。

2. コスト効率の高さ
   オープンソースライセンスのため、ベンダーロックインのリスクが低いです。必要に応じて違うクラウドに移行もより簡単になります。
   またクラウド上にコンテナ化してデプロイすることで、必要なリソースだけを利用・課金する運用形態も取りやすくなります。

3. 技術スタックの標準化
   アプリケーションやビッグデータ分析基盤で広く利用されているApache製品群を活用することで、エンジニアの学習コストや運用負荷を軽減できます。さらにKafkaやNiFiなどは他のクラウドサービスとも連携しやすく、データパイプラインを統合的に設計しやすいメリットがあります。

これらの理由から、Apacheコンポーネントを活用したCDCソリューションは、大規模かつリアルタイム性の高いデータパイプラインを比較的低コストで実装しやすく、将来的な拡張・変更にも柔軟に対応できる点で大きな魅力があります。

2. CDCソリューションにおけるApacheプロジェクトの概要

2.1 Apache Kafka

Apache Kafka は、分散メッセージングおよびストリーミングプラットフォームとして広く利用されているプロジェクトです、CDC ソリューションで中心的な役割を担うことが多いです。
Kafka はストリームデータのハブとしての役割を担い、その上にさまざまな処理系を組み合わせることで、リアルタイムなデータパイプラインを構築できるのが強みです。

2.2 Apache NiFi

Apache NiFi は、Web UIベースでデータフローを可視的かつ柔軟に設計・管理できる ETL/ELT プラットフォームです。CDCソリューションにおいても以下のように活躍します。

NiFi は、コードを書かずにバッチ/リアルタイム両方のデータパイプラインを実装できる点が最大の強みです。Debezium や Kafka と組み合わせて、CDC ベースのストリーミングフローを可視化・管理するケースが増えています。

2.3 Apache Flink

CDCソリューションで、リアルタイム分析や集計処理を本格的に行う場合、Apache Flink などの分散ストリーミングエンジンが役立ちます。

2.4 Apache Iceberg / Apache Kudu などのストレージ関連

CDCソリューションでは、最終的に変更イベントを蓄積・保存するストレージの選択肢も重要です。近年は「データレイク」や「レイクハウス」の概念が注目されており、以下のようなApacheプロジェクトが活発に利用されています。

Apache Iceberg

• テーブル形式のデータ管理を行うデータレイク用のテーブルフォーマット。
• ACID トランザクションやスナップショット管理など、データウェアハウスに近い機能をデータレイク上で実現。
• Incrementalな変更検出やタイムトラベルなど、CDCの利点を活かしやすい。
• Apache Kuduと比べてApache IcebergはHDFS上でも使えますし、クラウドストレージ（S3、ADLS Gen2など）でも使えます

Apache Kudu

• 低レイテンシ、かつアナリティクスにも適した列指向ストア。
• Hadoop エコシステムと統合しやすく、バッチ/ストリーム両方でリアルタイム読み書きが可能。
• データの更新・削除が扱いやすいため、CDC で流れてきた変更イベントを直接反映しやすい。
• Apache Icebergと比べてCompactionは全部Kuduが管理していますので、Compactionに関しての手間がないです

2.5 （補足）Debezium 等のオープンソースツール（※Apacheではないが連携が多い）

Debezium は、Apache Kafka を中心としたエコシステムで、さまざまなデータベースの変更データキャプチャを行うためのオープンソースツールです。Apacheプロジェクトではありませんが、Kafka Connect のコネクタとして動作することができます。

2.6 （補足）Oracle GoldenGate 等のツール（※Apacheではないが連携が多い）

Oracle GoldenGate は、エンタープライズ向けの CDC やデータレプリケーションのための商用ツールです。Apacheプロジェクトではありませんが、ビジネスクリティカルなシステムで広く利用されており、Kafka などのApache製品と連携させるケースも増えています。
GoldenGate で取得した変更イベントを Kafka に流し込み、その後 NiFi や Spark/Flink で処理するといったハイブリッド構成も多く見られます。企業の既存ライセンス環境や専門技術者のスキルセットを考慮しながら、Apache系ツールとの連携を図るのが一般的です。

3. システムアーキテクチャのパターン

3.1 Debezium + Kafka + NiFi を利用したアーキテクチャ例

この構成では、以下のようなCDC（Change Data Capture）のデータフローが実現されています。

• 1). Source Database (MySQL/PostgreSQL/MongoDBなど)
  アプリケーションやシステムで運用されているデータベースです。変更が行われるたびに、トランザクションログやオペレーションログなどに変更履歴が書き込まれます。
• 2). Debezium
  これらのデータベースの変更ログ（MySQLのbinlog、PostgreSQLのWAL、MongoDBのoplogなど）を監視し、データの変更内容をリアルタイムでキャプチャします。キャプチャした変更データはイベントとしてKafkaへ送信されます。
• 3). Apache Kafka
  Debeziumが生成した変更イベントを受け取るメッセージキュー/ストリーミングプラットフォームです。Kafka上のトピックに変更イベントが蓄えられるため、複数のコンシューマ（今回はNiFi）がこれを読み取って処理できます。
• 4). Apache NiFi
  Kafkaからの変更イベントを取り込み、必要に応じてルーティングや変換、フィルタリングなどの処理を行います。NiFiのフローファイルとして取り込んだデータを最終的にKuduへ連携します。
• 5). Apache Kudu
  NiFi経由で取り込まれたCDCデータを現行テーブルに反映する列指向ストレージエンジンです。
  分析処理やリアルタイムクエリに適した形でデータを蓄積できます。

全体としては、DBで発生した更新をDebeziumがいち早く検知し、Kafkaを介してNiFiが受け取り、最終的にKuduへ格納するというパイプラインになっています。これによりリアルタイムまたは近いリアルタイムでのデータ反映が可能になり、分析基盤として活用できる構成です。

3.2 Oracle GoldenGate + Kafka + NiFi を利用したアーキテクチャ例

上記 3.1 と似ていますが、データソースとして Oracle を使用し、CDC データは Oracle GoldenGate for Big Data から Kafka に送信されます。

NiFi は Kafka から CDC データを取得し、Kudu または Iceberg に書き込みます
（両方に書き込む必要はなく、どちらか一方で問題ありません）。

3.3 ストリーミング処理（Flink / Spark）を組み込む例

• 1). Source Database（MySQL/PostgreSQL/MongoDBなど）
  アプリケーションが利用しているデータベース。変更が起きると、トランザクションログやオペレーションログなどに履歴が書き込まれます。
• 2). Debezium
  Debezium を、Flink にライブラリとして組み込み、変更ログを直接キャプチャします。
  これにより、追加のMessage Queue（例えばApache Kafka）を挟まずに、データベースからの変更をリアルタイムで取得可能です。
• 3). Flink
  Flink は取り込んだ CDC（Change Data Capture）データのストリームを処理するためのプラットフォームです。
  Flink SQL などを用いて、リアルタイムにクエリを実行したり、データを変換・集計・フィルタリングすることができます。
• 4). Apache Kudu
  ストリーム処理した後の CDC データを格納する列指向ストレージです。
  最新のテーブル状態を反映できるため、分析やクエリを容易に実施できます。

4. NiFiでCDCデータを取ってくるときのデータフロー例

4.1 全体像

まずは全体図として、大きく３つのプロセッサーグループに分けています。

NiFiがKafkaから、CDCデータを取得し、その後KuduにInsert/Update/Deleteを反映します

4.2 CDCデータをApache Kafka への取り込み

• Topic 設計、パーティション設計

Kafka を使ったデータパイプラインで、高いスループットや効率の良い並列処理を実現するためには、Topic とパーティション数の設計が重要です。
ある程度まとまりの大きい単位（ドメイン単位、サービス単位、データ種類単位など）で Topic を分けておくと、メッセージやスキーマの管理が容易になります。たとえば、以下のように分類するイメージです。
• 正常系データ
• 異常系データ
• 監視データ

まずはこのレベルで分けておくことで、運用上の整理がしやすくなります。

さらに、正常系データの Topic をテーブル単位で分割しておくと、後々メッセージを探す際に見つけやすくなります。

• パーティション数の決め方

NiFi ノード数との関係: NiFi クラスターが複数ノードで構成されているので、Kafka のパーティション数が NiFi ノード数よりも十分に多くないと、並列処理のメリットが活かせません。
1 つのパーティションを 1 つのノードが担当するので、パーティション数 < ノード数 だと、あるノードは処理するパーティションが無い状態になり、負荷が均等に分散しないからです。

スループット要件: 取り扱うメッセージのレート（単位時間当たりの件数）に応じて、パーティション数を拡張しやすいようにしておく必要があります。高スループットが必要ならパーティション数を多めにしておくと、コンシューマグループ内で並列に処理ができるようになります。

アプリケーション側の制約: パーティションが多すぎると、メタデータ管理・リバランス時のオーバーヘッドも増加します。NiFi のスケールアウト計画や、運用管理にかけられるリソース（マシンスペックやネットワーク帯域など）とのバランスが大事です。

Kafka のパーティション数は後から増やすことはできるものの、再分割が走ることによるオーバーヘッドや、メッセージ順序の変化などの問題が発生し得ます。初期設計段階で余裕を持ったパーティション数を設定することが多いですね。

• Kafka Message Keyの決め方
  更新（つまり CDC データ）の多いテーブルと少ないテーブルが混在しているため、できるだけメッセージがパーティションに分散されるように、メッセージキーを選択する必要があります。
  例えばテーブルのPKをKafkaのメッセージキーにするのも良いです。

• スキーマ管理（Schema Registry の利用例）
  下記JSONをSchema Registryに登録。

{
  "name": "table_name_01",
  "type": "record",
  "namespace": "database_01",
  "fields": [
    {"name": "COLUMN01", "type": "string"},
    {"name": "COLUMN02", "type": "string"},
    {"name": "COLUMN03", "type": "int"},
    {"name": "COLUMN04", "type": ["null", "string"]},
    {"name": "COLUMN04_DATE", "type": ["null", {"type": "long", "logicalType": "timestamp-millis"}]}
  ]
}

4.2 KafkaからソースDBのCDCデータを取得 (NiFi Processors)

• Processor の選択：ConsumeKafka
  

ConsumeKafkaの設定値：

これでKafkaからデータをNiFiに取り込みます。

4.3 ストリーミング/バッチ処理 (NiFi)

• データクレンジング、フィルタリング
  関係ないデータを捨てる。
  また、必要なColumnを追加することができます。
• 重複排除
  KuduはUpsertができるので、同じPKのデータが来たら重複しないようにUpsertしたらできます。
  NiFi側で特別な操作をしなくても大丈夫です。

4.4 データストアへの書き込み (Kudu)

• パフォーマンスとレイテンシーのトレードオフ
  レイテンシー重視なら、OGGデータが来たらすぐKuduに反映することはできますが、Kudu書込がパフォーマンスのボトルネックになりやすいです。
  そのため、データをNiFiの中である程度纏めてからKuduに登録したほうが良いです。
  

キーポイントはMergeRecordプロセッサーを使い、同じop_typeのデータが来たら纏めてからKuduに登録します。

4.5 モニタリング・運用 (NiFi Monitoring / Kafka Monitoring / Grafana など)

NiFi のReporting Taskを活用します。
https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#:~:text=Reporting%20Task%3A%20Reporting%20Tasks%20run%20in%20the%20background,Reporting%20Tasks%20in%20the%20User%20Interface%20as%20desired.

例えば、PrometheusReportingTaskというReporting Taskがあります。

このコントローラーサービスを利用すれば、監視用のPrometheusに状態を送信することができます。

5. まとめ

本記事では、CDC（Change Data Capture）の概要と、そのソリューションとしてオープンソースの Apache 製品群を活用するメリット、さらに NiFi を中心とした具体的なデータフロー例を紹介しました。

具体的なポイントは以下のとおりです。

• 1). Kafka を利用した拡張性
  Kafka の Topic やパーティション数の設計を適切に行うことで、高スループットかつ分散処理が可能になります。NiFi クラスターのノード数や、更新頻度の高いテーブル・低いテーブルの混在などを考慮し、メッセージキー（例：テーブルの PK）を工夫してパーティションを効果的に活用します。
• 2). NiFi を利用した柔軟なデータフロー
  コード不要の GUI ベースで、JSON の加工やルーティング、バッチ処理とリアルタイム処理の両方に対応可能。データのクレンジングやフィルタリング、レコードのマージなどを組み合わせて、Kudu 等への書き込みを効率化できます。
• 3). ストレージの選択肢
  Kudu は低レイテンシで列指向のアップサートが得意なため、CDC データを直接反映しやすい利点があります。Iceberg はクラウドストレージ（S3、ADLS など）と組み合わせてデータレイクを構築しやすく、スナップショット管理や ACID 特性を活かした高度な分析にも向いています。
• 4). モニタリングとスキーマ管理
  NiFi の Reporting Task（例：PrometheusReportingTask）を活用すれば、フローの稼働状況やパフォーマンスを外部ツール（Prometheus, Grafana 等）で可視化可能です。また、Schema Registry を使うことでメッセージスキーマの互換性・バージョン管理を容易にし、Kafka と NiFi の連携をより安全・効率的に運用できます。

こうしたオープンソースの仕組みを組み合わせることで、大規模かつリアルタイム性の高いパイプラインを、比較的低コストかつ柔軟に構築できます。今後さらにデータ量の増大や処理要件が高度化しても、Apache コンポーネントを中心としたエコシステムを取り入れることで、拡張性や可用性を維持しながら運用していくことが可能になります。