データ取り込み（バッチ・CDC）｜データエンジニアリング

🎓 レベル：標準　|　重要度：A（必須）

📎 前提：ETLとELT　|　関連：べき等性と再実行・メッセージキューとイベントストリーミング

要点（BLUF）

取り込み（ingestion）は「源からデータ基盤へ引き込む」工程。方式は大きく 全件洗い替え（full）／増分（incremental）／CDC（変更データキャプチャ）。
増分は「前回以降に変わった分だけ」を取り込む。鍵はハイウォーターマーク（前回取り込んだ最大の更新時刻やID）を覚えておき、それより新しい行だけ引くこと。
CDCはDBのトランザクションログ（WAL/binlog）を読み、INSERT/UPDATE/DELETEをほぼリアルタイムに逐次捕捉する。源にクエリ負荷をかけず、削除も拾える。

概念 ── 全部引くか、差分だけ引くか

全件洗い替えは毎回テーブル全部をコピーします。単純で確実ですが、巨大テーブルでは源と回線に重い。増分は「前回からの差分」だけ引くので軽い反面、「何が変わったか」を正しく知る仕掛けが要ります。

flowchart TB
    F["全件（full）：毎回テーブル全部・単純だが重い"]
    I["増分（incremental）：更新時刻/ID > 前回値 の行だけ"]
    C["CDC：ログを読み INSERT/UPDATE/DELETE を逐次捕捉"]

仕組み ── ハイウォーターマークによる増分

「前回取り込んだ最大の updated_at」を覚え、それより新しい行だけを引きます。

-- 前回の到達点（watermark）より新しい行だけ抽出
SELECT id, updated_at, payload
FROM source_events
WHERE updated_at > :watermark
ORDER BY id;
-- 取り込み後、watermark を MAX(updated_at) に更新して次回へ

実行結果（実機・SQLite）:

初回ロード: 3 件取り込み, 新watermark=2026-06-02
増分ロード: 1 件取り込み（新着のみ）, 新watermark=2026-06-03
sink合計: 4 件

初回は全件、その後は新着1件だけが入り、合計4件。watermarkを進めながら差分だけ引くのが増分取り込みの基本骨格です。取り込みはUPSERTにしておくと、同じ行が再度来ても二重にならない（→ べき等性と再実行）。

設計の勘所 ── 方式の選択

方式	源への負荷	削除の捕捉	鮮度	実装
全件洗い替え	高い	自然に反映	低い（定期）	容易
増分（watermark）	低い	苦手（消えた行が分からない）	中	中（watermark管理）
CDC（ログ）	極小	得意（DELETEも拾う）	高い	高い（ログ連携・順序）

増分の弱点は削除：updated_atで「新しい行」は拾えても、「消えた行」は分からない。論理削除フラグを源に持つか、CDCにする。
CDC：DBの更新ログを読むため、源に対しては実質ゼロ負荷で、削除も順序付きで拾える。Debezium＋Kafka（→ メッセージキューとイベントストリーミング）が定番構成。

なぜそうするか ── 源を守りつつ最新を保つ

なぜ全件でなく増分・CDCを使うのか。源（本番DB）に負荷をかけず、かつ鮮度を上げるためです。本番DBに毎時フルスキャンを掛ければ業務が詰まります。増分なら差分だけ、CDCならログ読みで本番のクエリ系に触れずに済む。「源の健康を守る」のは取り込み設計の第一原則です。一方、増分は状態（watermark）を持つぶん壊れたときの復旧が要るので、べき等な再実行（→ べき等性と再実行）とセットで設計します。

⚠️ よくある落とし穴

watermarkに「処理時刻」を使う → 源の時計ずれ・遅延で取りこぼす。源のupdated_at等、データ自身の時刻を使う。
増分で削除を無視 → 基盤に幽霊レコードが残る。論理削除かCDCで対処。
updated_atが更新されないテーブル → 増分の前提が崩れる。トリガーやCDCで変更を確実に記録。
取り込みが非べき等 → 再実行で重複。UPSERT/MERGEで受ける。

対応ラボ

data-engineering-study/labs/04_etl_pipeline.py（PYTHONIOENCODING=utf-8 で実行・watermark増分を確認済み）。