メインコンテンツまでスキップ
に最終更新

Kafka と CDC の連携

要点 Apache Doris は、三つの経路で Kafka のトピックと CDC の変更ストリームをネイティブに取り込みます。FE 内で動作し、外部サービスを必要としない Kafka コンシューマーが欲しい場合は Routine Load を使います。MySQL/Postgres/Oracle のデータベース全体を、Stream Load 2PC による exactly-once で同期したい場合は、Flink CDC と Flink Doris Connector を使います。Kafka Connect がすでに稼働しており、ペイロードが Avro、Protobuf、または Debezium の場合は Doris Kafka Connector を使います。どの経路でもデータは同じ Unique Key Merge-on-Write テーブルに書き込まれるため、到着順が前後するイベントや削除は、選んだ経路にかかわらず同じように処理されます。

Apache Doris の Kafka と CDC の連携: Routine Load で Kafka のトピックを Doris に直接取り込むか、Flink を通じて MySQL/Postgres の CDC データを Stream Load 2PC による exactly-once でストリーミングします。

Apache Doris で Kafka と CDC の連携を使う理由​

Apache Doris の Kafka と CDC の連携では、ストリーミングによる取り込み、exactly-once 配信、到着順が前後するデータの処理をエンジン内で行えるため、トピックや binlog からクエリ可能な行までの経路を短くできます。多くのデータウェアハウスでは、ストリーミング取り込みの仕組みを別途追加する必要があります。Kafka コンシューマーを立て、オフセット管理、再試行、スキーマのマッピング、削除処理を実装し直したうえで、データウェアハウスに障害が起きたときにコンシューマーが正常に停止することを期待しなければなりません。あるいは、前段で Flink を動かして追加のクラスターを維持します。または「リアルタイム」を諦め、一括データロードで五分ごとのバッチを送ることになります。

  • CDC パイプラインがチェックポイント失敗後に直前のバッチを再実行し、バッファリングしていたすべての行を重複させてしまう。
  • Unique Key テーブルで到着順が前後した更新が古いバージョンに負けてしまう。どちらの書き込みが新しいかをエンジンに伝える情報がないためです。
  • Avro または Protobuf を含む Kafka トピックを、JSON と CSV しか扱えない HTTP データロード API に取り込めない。

Doris はこうした処理の大部分をエンジン内で担います。そのため、Kafka のトピックや MySQL の binlog からクエリ可能な行までの経路が短くなり、障害の原因も把握しやすくなります。

Apache Doris の Kafka と CDC の連携とは​

Apache Doris の Kafka と CDC の連携は、共通の基盤を使うストリーミングおよび CDC のデータ取り込み経路の集合です。その基盤を構成するのは、Label で識別されるトランザクション単位のデータロード、upsert と削除を扱う Unique Key Merge-on-Write テーブル、そしてイベントの到着順が前後したときに採用する行のバージョンを決めるシーケンス列です。経路ごとに異なるのは、コンシューマーをどこで動かすかです。

主な用語

  • Routine Load: FE が管理する、継続的に動作する Kafka コンシューマーです。CREATE ROUTINE LOAD 文を一つ実行すると、FE が BE にタスクを割り当てます。BE はバッチを取得し、各バッチをトランザクション単位のデータロードに変換します。
  • Stream Load 2PC: 後続の呼び出しで txn_operation: commit または txn_operation: abort が送られるまで、PRECOMMITTED で止まる HTTP データロードです。この二段階の分離により、外部の sink で exactly-once を実現できます。
  • Flink Doris Connector: Stream Load 2PC を実行し、commit を Flink のチェックポイントと連動させる外部の Flink sink です。Flink CDC と組み合わせると、データベース全体を同期し、DDL を伝播できます。
  • Doris Kafka Connector: Kafka Connect の sink プラグインです。Kafka Connect の converter エコシステムを利用できるため、独自のシリアライザーを用意せずに Avro、Protobuf、Debezium のペイロードを扱えます。
  • function_column.sequence_col: Unique Key MOW テーブルのテーブル単位のプロパティで、行のどのバージョンを採用するかを値によって決める列を指定します。到着順が前後する CDC データへの安全策です。

Apache Doris の Kafka と CDC の連携の仕組み​

Apache Doris の Kafka と CDC のどの取り込み経路でも、データが FE に届いた後の流れは共通です。バッチを取得し、Label 付きのトランザクションにまとめ、確認を受けて commit し、すべてのレプリカに公開します。

  1. コンシューマーがバッチを取得します。 Routine Load は BE にタスクを割り当て、Kafka から max_batch_rows（デフォルトは 20M）、max_batch_size（デフォルトは 1 GB）、max_batch_interval（デフォルトは 60 s）のいずれかに最初に達するまでデータを取得します。Flink CDC は、チェックポイントの処理中も開いたままのチャンク形式の HTTP Stream Load を使って行をストリーミングします。
  2. バッチがトランザクションになります。 BE は Label を付けてトランザクションを開始します。Routine Load は Label を生成し、Flink Doris Connector はチェックポイント ID を付けます。いずれの場合も、同じ Label で再送されたデータは重複排除されます。
  3. 確認を受けてから commit します。 Routine Load はバッチの取り込みが完了すると commit し、成功した場合にのみ Kafka のオフセットを進めます。Flink CDC は PRECOMMITTED で止まり、次のチェックポイントバリアで txn_operation: commit を実行します。チェックポイントに失敗した場合は abort が実行され、準備済みのデータが破棄されます。
  4. 公開すると行が参照可能になります。 バックグラウンドのデーモンが各レプリカに PublishVersion を送ります。トランザクションが VISIBLE に達すると、新しい行をクエリできます。同じトランザクション内のタブレットが、部分的にだけ反映されることはありません。
  5. エラーが起きるとジョブが一時停止し、データは維持されます。 Routine Load のバッチで max_error_number（デフォルトは 0）を超えると、ジョブは PAUSED になり、FE に最後のエラー URL が保持されます。ソースデータを修正してから再開してください。オフセットは元の位置にとどまります。

クイックスタート​

CREATE TABLE events (
  user_id BIGINT, event_type STRING, ts DATETIME
)
DUPLICATE KEY(user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 8;

CREATE ROUTINE LOAD events_kafka ON events
COLUMNS(user_id, event_type, ts)
PROPERTIES (
  "format" = "json",
  "jsonpaths" = "[\"$.user_id\",\"$.event_type\",\"$.ts\"]"
)
FROM KAFKA (
  "kafka_broker_list" = "localhost:9092",
  "kafka_topic" = "events",
  "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

想定される結果

+--------+--------------+---------+----------------------+
| Id     | Name         | State   | Statistic            |
+--------+--------------+---------+----------------------+
| 100123 | events_kafka | RUNNING | {"loadedRows":4821,..|
+--------+--------------+---------+----------------------+

SHOW ROUTINE LOAD FOR events_kafka を実行すると、ジョブの状態が RUNNING であること、パーティションごとの最後に消費したオフセット、ロードされた行数を確認できます。PAUSE、RESUME、STOP は、他の Doris ジョブと同じように使えます。

CDC の場合は、コンシューマーの代わりに Flink CDC と Flink Doris Connector を使います（Flink ジョブ一つ、MySQL source、Doris sink、sink.enable-2pc = true）。スキーマとシーケンス列は Doris 側で定義します。

CREATE TABLE orders (
  order_id BIGINT, status STRING, updated DATETIME
)
UNIQUE KEY(order_id) DISTRIBUTED BY HASH(order_id) BUCKETS 8
PROPERTIES ("function_column.sequence_col" = "updated");

sequence_col の設定により、新しい CDC イベントの後から遅延したイベントが届いても、行が保護されます。古い updated の値は採用されず、新しい値が採用されるため、到着順にかかわらず結果は同じです。

Apache Doris の Kafka と CDC の連携を使う場面​

ストリーミングイベントや上流データベースの変更を、五分ごとのバッチ処理を待たず、exactly-once 配信と正しい順序で Doris に取り込む必要がある場合に使います。

適しているケース

  • CSV または JSON を含む Kafka トピックを、追加のサービスを使わず Doris に取り込みたい場合。Routine Load が最短の経路です。
  • MySQL、Postgres、Oracle の CDC データを、exactly-once 配信と DDL の伝播を伴って Doris に取り込みたい場合。Flink CDC と Flink Doris Connector を使います。
  • Kafka Connect がすでに稼働しており、ペイロードが Avro、Protobuf、Debezium の場合。Doris Kafka Connector なら形式の不一致を解消できます。
  • Unique Key テーブルを使ったリアルタイムダッシュボードで、到着順が前後する書き込みも正しい順序で処理する必要がある場合。上記のどの経路でも function_column.sequence_col を組み合わせます。

適していないケース

  • Routine Load を使って Kafka から Avro を取り込む場合。Routine Load が読み取れるのは CSV と JSON だけです。上流で形式を変換するか、適切な converter を設定した Doris Kafka Connector を使ってください。
  • マイクロサービスからイベントごとに INSERT INTO ... VALUES を実行する場合。これは Kafka や CDC で扱う問題ではありません。Group Commit を使うと、BE 側で小さな書き込みをまとめられます。
  • Kafka の保持期間が短いトピックで Routine Load を使う場合。保持期間が 24 h でジョブが二日間一時停止すると、保存済みのオフセットが取得可能な最も古いオフセットより前になり、Kafka は再開を "out of range" として拒否します。保持期間を延ばすか、データをバッファリングできる connector を使ってください。
  • FE だけでオフセットの遅延を監視する場合。FE は現在、遅延のメトリクスを公開していません（未解決の issue）。Kafka 側の consumer-lag exporter からメトリクスを収集するか、BE 側の処理速度を監視してください。
  • クラウド／ストレージ・コンピュート分離構成の Merge-on-Write Unique Key テーブルに対して Stream Load 2PC を使う場合。BE はこの組み合わせを拒否するため、その構成では Flink Connector の exactly-once 経路が Label による重複排除とシーケンス列の組み合わせに切り替わります。オンプレミスのクラスターでは、MoW に対して Stream Load 2PC の経路をそのまま使えます。

パフォーマンスと数値​

2022 年七月のリリースブログ「Flink による Apache Doris へのリアルタイム書き込みが、高速性と exactly-once を両立する仕組み」によると、Stream Load の同時実行数が 30 未満のとき、エンドツーエンドのレイテンシーは一秒未満でした。20 個の Flink タスクで、上流から毎秒約 100,000 イベントを継続して処理できました。また、高い同時実行数では、以前のデータロード経路と比べて BE の CPU 使用量が約 25% 減少しました。出典: Apache Doris ブログ。

関連資料​

Report issue