月曜の朝、営業部長から連絡が来ます。「在庫ダッシュボードの数字が、また昨日のままなんだけど」。基盤担当のあなたは、日次バッチをやめて業務DBの変更をそのまま流す仕組みを入れたいと考えています。しかしDBAは「本番DBに何かをつなぐのは困る」と渋り、情シスは「Kafkaなんて誰が面倒を見るのか」と聞いてきます。経営企画からは「それで月にいくら増えるのか」と確認が入ります。

DBの変更を行単位で拾って、別の場所へ順に流す仕組みをCDC（Change Data Capture、変更データキャプチャ）と呼びます。CDCの道具は似た名前で並んでいますが、得意な場面はまったく違います。どこからどこへ流すのか、何分の遅れまで許されるのかを先に決めれば、候補はほぼ1つに絞れます。

流す元と流す先で、候補はほぼ決まる

状況第一候補これを選ぶと後で困ること
MySQLやPostgreSQLなどの業務DBから、DWHや複数の下流へ秒〜分単位で流したいDebezium（Kafka Connect上、またはDebezium Server）Kafka Connectかメッセージング基盤の運用が増える。binlog・WALの保持設計を誤ると取りこぼす
送り手も受け手もPostgreSQLPostgreSQLの論理レプリケーションDDL（テーブル定義の変更）とシーケンスの値は複製されない。使っていないスロットがWALを溜め続ける
Snowflakeに入ったテーブルの差分だけを、下流のテーブルへ反映したいSnowflake Streams＋Tasks、またはDynamic Tables外部DBの変更は拾えない。ストリームを読む処理が止まると、変更履歴が期限切れになる
数十分〜1時間の遅れは許され、運用の手間を最小にしたいマネージドの取り込みサービス（Fivetran、AWS DMSなど）件数や行数に応じて費用が増える。コネクタの挙動を細かく変えにくい
1日1回で足り、物理削除がほとんどない更新日時での差分取得（クエリ方式）のまま物理削除した行を検知できない。更新日時を更新しない処理があると漏れる

最後の行は見落とされがちです。WHERE updated_at > 前回時刻 で差分を取る方式は、DBから消えた行を見つけられません。消えた行は、もう問い合わせても返ってこないからです。削除を下流に伝えたいなら、DBの変更ログを読む方式（ログベース）が要ります。

取り込みツール全体の比較はデータ取り込みツールの選び方に、バッチとストリーミングのどちらにするかはバッチとストリーミングの選び方にまとめています。

業務DBから流すならDebezium：先にDB側の設定を確かめる

DebeziumはDBの変更ログを読み、行ごとの変更をイベントとして流すOSSです。2026年10月時点の安定版は3.7で、安定版のコネクタはMySQL・MariaDB・PostgreSQL・SQL Server・Oracle・Db2・MongoDB・Cassandraに対応しています。

MySQLとAmazon RDSでは、binlogの保持時間を最初に確かめる

MySQLから読む場合、公式ドキュメントは binlog_format = ROW と binlog_row_image = FULL を求めています。接続ユーザーには SELECT・RELOAD・SHOW DATABASES・REPLICATION SLAVE・REPLICATION CLIENT の権限が要ります。

DBAとの打ち合わせで真っ先に確認したいのは、binlog（変更ログ）が何時間残るかです。MySQL本体の既定は30日（binlog_expire_logs_seconds = 2592000）です。しかしAmazon RDS for MySQLでは事情が違います。binlog retention hours の既定値は NULL で、これは「保持しない（0時間）」を意味します。設定できる上限は168時間（7日）です。

-- RDS for MySQL：binlogを72時間残す（上限は168時間）
CALL mysql.rds_set_configuration('binlog retention hours', 72);
-- 現在の設定を確かめる
CALL mysql.rds_show_configuration;

連休中にDebeziumが止まり、その間にbinlogが消えれば、続きから読む手段はありません。最初から全件を取り直すことになります。保持時間は「Debeziumが止まっても、気づいて直すまでに要る時間」から逆算して決めます。

初回の全件取り込みと、届くイベントの形

コネクタは最初に既存データを読み込みます。PostgreSQLコネクタの snapshot.mode の既定は initial で、オフセット（どこまで読んだかの記録）がまだ無いときに全件を読みます。大きなテーブルは、シグナル用テーブルを使う増分スナップショットで分割して読めます。既定では1回あたり1,024行ずつです。

下流に届くイベントには before（変更前）、after（変更後）、op（操作の種類）が入ります。

op意味下流での扱い
c挿入そのまま追加
u更新主キーで突き合わせて上書き
d削除物理削除するか、削除フラグを立てるかを先に決める
rスナップショットでの読み込み挿入と同じ扱いにすることが多い
tTRUNCATE（PostgreSQL）下流のテーブルも空にするか決める

d の扱いを決めずに作ると、消したはずの受注が集計に残り続けます。

Kafkaが無くても使える

Debeziumを使うにはKafkaが必要だと思われがちです。しかしDebezium Serverという単体アプリを使えば、Amazon Kinesis、Google Cloud Pub/Sub、Apache Pulsar、Azure Event Hubs、Redis、HTTPなどへ直接送れます。社内にKafkaの運用者がいないなら、まず既に使っているクラウドのメッセージングへ送る構成を検討します。

送り手がPostgreSQLなら、レプリケーションスロットを見張る

PostgreSQLで変更を外へ取り出すには wal_level = logical が要ります。DebeziumもPostgreSQL同士の論理レプリケーションも、PostgreSQL 10以降に標準で入っている pgoutput で変更を読めます。

変更を読む側には、レプリケーションスロット（読み終わるまでWALを消さずに残す目印）が1つずつ作られます。読み手が止まっても取りこぼさずに済むのは、このスロットのおかげです。その代わり、読み手が止まったままだとWALは消えずに溜まり続け、送り手のディスクを圧迫します。

対策は3つあります。1つ目は、PostgreSQL 13以降の max_slot_wal_keep_size でスロットが抱えるWALの上限を決めることです。既定は -1（無制限）です。上限を超えると読み手は続きから読めなくなります。それでも、本番DBのディスクが埋まるよりは被害が小さく済みます。2つ目は、PostgreSQL 18で入った idle_replication_slot_timeout です。一定時間使われていないスロットを無効にでき、既定は 0（無効）です。3つ目は、スロットが抱えているWALの量を定期的に見ることです。

-- スロットごとに、溜まっているWALの量を確かめる
SELECT slot_name,
       active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

Debeziumのドキュメントには、もう1つ落とし穴が書かれています。同じPostgreSQLインスタンスに、更新の多いDBと更新の少ないDBが同居している場合です。WALはインスタンス全体で共有されるので、取り込み対象のDBに変更が来ない間もWALは溜まります。heartbeat.interval.ms と heartbeat.action.query で定期的に小さな書き込みを起こし、読んだ位置を進めさせます。

削除イベントの中身にも注意が要ります。テーブルの REPLICA IDENTITY が既定（DEFAULT）のままだと、削除や更新の before に入るのは主キーの値だけです。変更前の全列を下流で使いたいテーブルは FULL にします。主キーの無いテーブルを既定のまま使うと、削除イベントに before 自体が入りません。

PostgreSQL同士の複製では、DDLとシーケンスの値が複製されない点も押さえておきます。列を追加するときは、受け手のテーブルを先に変えておくとエラーを避けやすくなります。設定の手順とPublication・Subscriptionの書き方はPostgreSQL論理レプリケーションで詳しく扱っています。

Snowflakeの中の差分処理は、Streamsの「消費」の仕組みを理解してから

Snowflake Streamsは、テーブルの変更を記録する目印です。ストリームを読むだけでは目印（オフセット）は進まず、INSERTやMERGEなどのDMLの中で使ったときに初めて進みます。この性質を知らないと、テストで SELECT しただけの差分が本番の処理で消えずに残り、二重に反映される原因になります。

種類追跡する変更使える主な対象
Standard挿入・更新・削除・TRUNCATE通常のテーブル、Dynamic Tables、ビューなど
Append-only挿入だけ通常のテーブル、Dynamic Tables、ビューなど
Insert-only挿入だけ外部テーブル、外部管理のIcebergテーブルなど

更新は METADATA$ACTION が DELETE と INSERT の2行の組で届き、METADATA$ISUPDATE が TRUE になります。MERGEで反映するときは、この組を1行として扱います。

CREATE TASK task_sync_orders
  WAREHOUSE = WH_TRANSFORM
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('stream_orders')
AS
MERGE INTO gold.fct_orders t
USING (
  SELECT * FROM stream_orders
  WHERE NOT (METADATA$ACTION = 'DELETE' AND METADATA$ISUPDATE)  -- 更新の「旧行」は除く
) s
ON t.order_id = s.order_id
WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' THEN DELETE
WHEN MATCHED THEN UPDATE SET t.amount = s.amount, t.status = s.status
WHEN NOT MATCHED AND s.METADATA$ACTION = 'INSERT' THEN
  INSERT (order_id, amount, status) VALUES (s.order_id, s.amount, s.status);

ALTER TASK task_sync_orders RESUME;  -- タスクは作成時に停止状態

運用で詰まるのは次の3点です。1つ目は、1つのストリームを2つの処理で読むと、先に読んだ側だけが差分を受け取ることです。読み手ごとにストリームを作ります。ストリームが持つのはオフセットだけなので、複数作ってもストレージはほとんど増えません。2つ目は期限切れです。読まれないストリームのために、Snowflakeは元テーブルの保持期間を既定で最大14日まで延ばします。それを過ぎるとストリームは古くなり、作り直しになります。3つ目は費用です。自分で指定したウェアハウスで動くタスクは、起動のたびに最低60秒分の課金があります。

同じことを宣言的に書けるのがDynamic Tablesです。SELECT文と目標の遅れ（TARGET_LAG、最小60秒）を書けば、Snowflakeが更新を引き受けます。

やりたいこと向く道具
SELECTで書ける変換を、手順を書かずに最新に保ちたいDynamic Tables
削除の反映や外部への通知など、更新の手順を自分で書きたいStreams＋Tasks
dbtプロジェクトでテストやドキュメントと一緒に管理したいdbtのincrementalモデル

事故になりやすい3つの場面と、その費用

症状原因事前の手当て
再起動したら続きから読めないbinlog・WALが先に消えた保持時間を復旧に要る時間から逆算する。RDSは既定で保持しない
消した行が集計に残る削除イベントを下流で無視しているd と METADATA$ACTION = 'DELETE' の扱いを設計書に書く
列を足したらパイプラインが止まるスキーマの変更が下流に届かない列の追加は受け手を先に変える。互換性のルールを決めてから変える

1つ目の手戻りがどれくらい重いか、例で試算します。年収700万円のエンジニアの会社負担は、給与の約3倍で年2,100万円です。年1,900時間で割ると、時給は約11,000円になります。binlogが消えて全件を取り直し、下流の数字と突き合わせるのに2人で3日（48時間）かかれば、人件費だけで約53万円です。その間、ダッシュボードも止まります。金額はいずれも説明のための例です。

自前で運用するか、マネージドに任せるか

選択肢向く状況確かめること
Debezium＋Kafka Connectを自社運用Kafkaの運用者がいて、同じ変更を複数の下流へ配りたい障害時に誰が起きるか。binlog・WALの保持とスロットの監視
Confluent CloudのDebeziumベースのコネクタKafkaは使いたいが、コネクタの運用は持ちたくないPostgreSQL CDCはV2コネクタに移行済みか（V1は提供終了）
AWS DMSAWS上のDBから、全件移行と継続的な差分の取り込みをまとめて済ませたい取り込み元のDB種別と版が対応しているか
Fivetranなどの取り込みサービス1本のDBからDWHへ、運用の手間をかけずに入れたい件数・行数に応じた費用。削除の扱いの既定

自前で運用するかどうかは、ツールの機能より「夜中に止まったとき誰が直すか」で決まります。答えられないうちは、マネージドから始めるほうが安全です。

よくある質問（FAQ）

CDCを入れると、業務DBは重くなりますか?

ログを読む方式は、テーブルへ定期的に問い合わせを投げません。ただし負荷がゼロになるわけではありません。binlogやWALを長く残す分だけディスクを使い、初回の全件取り込みでは大きな読み込みが走ります。全件取り込みは業務時間外に回すか、増分スナップショットで分割して読みます。

更新日時で差分を取る方式では、何が足りないのですか?

物理削除を検知できない点です。消えた行は問い合わせても返ってこないので、下流には残り続けます。更新日時を更新しない一括処理があると、その変更も漏れます。削除がほとんど無く、1日1回で足りるテーブルなら、この方式のままでも構いません。

Debeziumを使うには、Kafkaを立てる必要がありますか?

必須ではありません。Debezium Serverを使えば、Kinesis、Pub/Sub、Pulsar、Event Hubs、Redis、HTTPなどへ直接送れます。

下流で同じ変更が二重に届くことはありますか?

あります。障害からの再開時には、同じイベントがもう一度届くことを前提に作ります。下流の反映は主キーでのMERGEにして、同じイベントを2回適用しても結果が変わらないようにしておきます。

SnowflakeのStreamsとDynamic Tablesは、どちらを使えばよいですか?

SELECT文で書ける変換ならDynamic Tablesが手軽です。目標の遅れを書くだけで、更新の手順はSnowflakeが受け持ちます。削除の反映方法を細かく決めたいときや、変更をきっかけに外部へ通知したいときは、Streams＋Tasksで手順を書きます。1分より新しいデータが要る場合、Dynamic Tablesは向きません。

出典

「本番DBに何かつなぐのは困る」と言われたまま止まっているCDCの設計も、binlogやWALの保持時間の決め方から一緒に整理できます。初回相談（30分・無料）で一緒に整理できます。