ブログ一覧

ETLの差分更新でデータが欠ける・二重になる原因：CDC・再実行・品質検証の設計

2026年10月27日 · 約10分で読めます

ジョブは成功したのに、売上が合わない。更新日時のカーソルが落とす遅いコミット、CDCの初期取得と削除、再実行時の公開、JOIN後の金額、鮮度と完全性を、受注データの分析基盤を例に5件のTipsから解説します。

ETLの差分更新とは？成功したジョブと、合わない売上

差分更新は、前回から変わったデータを取り込み、保存先へ反映する方法です。ETLは抽出・変換・ロード、ELTはロード後に変換する構成を指します。どちらでも、欠損と重複を防ぐには、抽出する境界、再開位置、公開する単位を決める必要があります。

最初に調べるのは、元の注文が取り込まれたか、再実行で二重に公開されていないか、集計が注文単位を保っているかです。ジョブの終了コードだけでは、この3つの答えは分かりません。5件のTipsを使って、受注DBから売上画面まで順に追います。

updated_atで取りこぼす理由：更新日時とコミット順は違う

前回の最大更新日時より新しい行だけを読む方式を考えます。PostgreSQL 18のRead Committedでは、通常のSELECTは文の開始前にコミットしたデータを読みます。この可視性の仕様から、コミット前に付いた時刻をカーソルにすると、次の取りこぼしが起きると推論できます。

IDを組み合わせても、Aの古い時刻は変わりません。並び順を一意にすることと、後から可視になる変更を捕まえることは別です。連番IDだけを使う案も、採番順がコミット順と一致するとは限らないため、同じ確認が必要です。

SELECT id, updated_at, amount_yen, status
FROM orders
WHERE (updated_at, id) > ($1, $2)
  AND updated_at < $3
ORDER BY updated_at, id
LIMIT 1000;

updated_atはNOT NULL、idは一意かつNOT NULLとし、$1と$2は前ページ末尾の時刻とID、$3は今回固定した抽出上限です。値はバインドします。この行比較は同時刻の行を飛ばさずページングするためのものです。固定した可視データ内の説明であり、実行中に行が更新される抽出全体や、遅いコミットを含む欠損防止を、このSQLだけで保証するものではありません。

更新日時で続けるなら、すべての更新経路で時刻を更新する契約を確認し、一定範囲を重ねて読み直します。重複は主キーと信頼できる版で処理し、古い値で新しい値を上書きしないようにします。ただし有限の再読込期間では、その期間を超える遅れを救えません。物理削除した行もSELECTには現れないため、削除記録や定期照合などが必要です。

差分抽出・CDC・全量更新をどう選ぶ？

欲しいものが現在の状態なのか、途中の変更履歴までなのかを決めます。注文が2回更新されても、ポーリング時に残るのは最後の状態かもしれません。今回の現在状態の集計と、すべての状態遷移を監査する用途では、同じ方式を無条件に採用できません。次の比較は、この受注サービスで判断するための整理です。

CDCの初期取得：コピーの終了時刻からログを読まない

CDC（Change Data Capture）は変更を取り出す仕組みです。DebeziumのPostgreSQLコネクターは、初期の整合したスナップショットを取得し、それに対応するWAL（Write-Ahead Log）の位置から変更の読み取りへ移ります。コピー中の変更を覆う境界が必要で、コピーが終わった時計の時刻を任意に開始位置へ使うわけではありません。

CDCを使っても受け口の設計は残ります。復旧時には同じ変更が再送され得るため、単純に売上へ加算すると二重計上します。今回なら注文IDごとの現在状態へ適用し、適用順と版を管理します。削除も明示的に反映します。複数表を一貫して集計するなら、元のトランザクションの完了を含む適用済み境界を定めます。チェックポイントは保存先への反映を耐久化してから進め、別システムに置く場合は、その間の停止で再送されても結果が壊れないようにします。

PostgreSQLのレプリケーションスロットは、読み手が必要とするWALを保持する経路を持ちます。停止中の保持量とディスク余力を監視し、保持の上限やスロットの有効性も確認します。必要なログを失った場合は、単に最新位置へ飛ばすと欠損が残ります。再スナップショットと照合を行い、検証が済むまで分析画面を確定値として更新しない復旧手順を決めます。

再実行で二重にしない：ステージングから検証後に公開する

10月1日分の集計100行を書いた直後、完了応答だけが失われたとします。再実行で100行を追記すると200行になります。新しい実行IDを付けても、同じ業務データを重ねて保存した事実は変わりません。Airflowのベストプラクティスも、再試行で結果を変えないため、入力範囲の固定やUPSERTなどを勧めています。

ここでは利用者から見えないステージングへ対象期間を作り直し、検証後にまとめて公開します。再現に必要なのは期間だけでなく、元データをどの時点まで取り込んだ版かと、変換コードの版です。毎回「最新」の入力を読むと、再試行の間にデータが変わり、同じ計算になりません。

expected_generation = read_current_generation(interval)
input = pin_input_version(interval, source_cut)
output = build_in_staging(input, transform_version)
validate(output, expected_scope)

with destination_transaction():
    lock_publication_record(interval)
    if committed(publication_id):
        result = recorded_result(publication_id)
    else:
        require current_generation(interval) == expected_generation
        replace_interval(interval, output)
        record_publication(publication_id, input.version, transform_version)
        advance_generation(interval)
        result = publication_id
return result

publication_idは同じ公開操作の再試行で固定します。対象期間の公開管理行をロックし、実行記録があれば保存済みの結果を返します。未公開なら、ステージング作成前に記録したexpected_generationと現在の世代が一致する場合だけ、出力・公開記録・世代を同じトランザクションで変更します。競合ならロールバックし、現行版を確認して再作成するか停止します。疑似コードのwithは正常終了時にコミット、例外時にロールバックする抽象です。公開IDの一意性も保存先で強制します。

コミットの応答が失われたら、再接続後に同じpublication_idの記録で結果を照会します。これは単一の保存先DBで、出力と公開記録を一緒にコミットする設計です。PostgreSQLのトランザクションは途中の変更を他のトランザクションへ見せません。ただし複数クエリで読む画面では、1つの版を固定して読む契約も必要です。外部ファイルへの書き込みまで、このDBのトランザクションでまとめられるとは扱いません。

オブジェクトストレージなら、完成した不変のファイル群を指すマニフェストを作り、参照先を条件付きで切り替える方式などを検討します。読み手もその公開契約に従います。複数ファイルを順に移動するだけでは、全体の一括公開にはなりません。古い実行が後から完了しても、新しい公開版を上書きできない条件を保存先側で強制します。

件数が合っても金額が違う：JOIN後の粒度を確認する

取り込みと公開が正しくても、変換で金額が増えることがあります。粒度は「1行が何を表すか」です。注文1件の3,000円を明細3行へJOINすると、各行に3,000円が載り、単純なSUMは3,000 × 3 = 9,000円になります。3行あることは正しくても、注文単位の売上は3,000円です。

SUM(DISTINCT amount_yen)では一般的に直せません。3,000円の別注文が2件あれば、正しい6,000円まで3,000円に縮めてしまいます。注文単位で集計するか、明細金額を業務ルールで定義してから合計します。全体の件数も、欠損1件と重複1件が相殺すれば一致してしまいます。

dbtはunique・not_null・accepted_values・relationshipsと独自のデータテストを提供します。ただし売上の意味や照合する範囲は利用側で決めます。元DBの現在値と昨日の確定版をそのまま比べず、締め時点を揃えます。毎回は変更範囲を確認し、定期的に全体を照合する案なら、全体の異常を見つけるまでの遅れも運用に含めます。

鮮度と完全性：最新の1行が届いても、範囲が揃ったとは限らない

09:45にジョブが成功し、最新の1行には09:44が付いている。それでも、連続して揃い、売上画面へ公開済みと確認できたデータが09:00までなら、確認済み範囲の遅れは45分です。15分以内という独自の目標に対して30分超過しています。最新1行だけを見る計算の1分とは、測っているものが違います。

dbtのfreshnessは指定した時刻列や対応するメタデータなどで鮮度を検査します。その検査だけで全区間の完全性が分かるとは扱いません。この設計では入力元ごとに完了した連続範囲を記録し、すべての必須入力が揃った範囲だけを公開可能とします。正常な0件と入力停止は、営業日カレンダーや生存確認・完了通知で区別します。

設計を確かめる：停止・再送・欠損を含むテスト

正常な1回の実行だけでは、再開できることは確かめられません。採用した方式に応じて次の障害を起こし、保存先の実データと利用者が読める版を確認します。これはこの分析基盤のための試験案で、この記事で実施した製品検証ではありません。

1. カーソルより古いupdated_atを持つ変更を、抽出後にコミットする。再読込の期間内外で、検出・修復の違いを確認する。
2. 初期コピー中に注文を更新・削除する。CDCへの引き継ぎ後、同じ締め時点の元データと保存先を照合する。
3. 保存先への反映直後、再開位置の保存前に停止する。同じ変更が再送されても金額が二重にならず、削除済みの行が復活しないことを確かめる。
4. 公開前とコミット応答の直前で停止する。未検証の版が読めず、同じ公開IDで結果を照会・再試行できることを確かめる。古い実行を遅れて完了させ、世代の競合で止まることも試す。
5. 明細が複数の注文、同額の別注文、欠損と重複が相殺する入力を用意し、件数以外の検査が働くことを確認する。
6. 一部の入力だけ止め、別の入力から新しい1行を届ける。完全性の通知が消えないことと、正常な0件を区別できることを確かめる。

よくある疑問：短い間隔・UPSERT・CDCだけで解決する？

• 抽出間隔を短くすれば欠損を防げる？ 観測の遅れは減らせても、古い更新日時で後からコミットする順序は直りません。
• UPSERTなら再実行は安全？ 一意キーでの二重行は防げても、古い版による上書き、削除、集計への二重加算、公開途中の参照は別に設計します。
• CDCならend-to-endのexactly-onceになる？ 取得・配送・保存先・集計・公開の境界を含めた保証を確認します。再送が起こり得る構成では、受け口が同じ変更を安全に扱う必要があります。

参考にした一次資料

以下は2026年10月27日に確認した公式資料です。製品の仕様と、仕様を使って組み立てた独自の障害例・設計案を区別しています。導入時には、採用するコネクター・保存先・アダプターの版と設定に合わせて確認します。

• PostgreSQL 18 — Transaction Isolation: Read Committed
• PostgreSQL 18 — Row Constructor Comparison
• Debezium — PostgreSQL connector: snapshots and recovery
• PostgreSQL 18 — Logical Decoding Concepts: replication slots
• Apache Airflow — Best Practices: creating a task
• PostgreSQL 18 — Transactions
• dbt — Data tests
• dbt — Freshness configuration