ETLのベストプラクティスは、原則の一覧として知っていても、コードに落とした瞬間に崩れます。再実行したら行が倍になった、遅れて届いた注文だけ欠けた、認証エラーを朝まで再試行し続けた、という事故の多くは書き方の細部で起きるからです。この記事では、ETLの仕組みやELTとの違いを押さえた実装者向けに、冪等な書き込み、増分ロード、リトライ、データ品質チェック、監視の5つを、PostgreSQL 18とPythonで動くコードで示します。後半ではAirflowとdbtへの載せ替え方と、小規模なETLで省いてよい項目も扱います。
まとめ：ETLのベストプラクティスは再実行しても壊れない書き込みから決める
優先順位をはっきりさせておきます。最初に決めるのは書き込み方です。ステージング表に入れてから主キーでUPSERTする形にしておけば、リトライも再実行もバックフィルも「もう一度流すだけ」で済みます。ここが決まらないうちにリトライを足すと、再試行のたびに重複が増えるだけです。
2番目が増分ロードで、前回の到達点をウォーターマーク表に持ち、そこから一定時間さかのぼって読み直します。3番目がリトライで、接続断やタイムアウトのような一時障害だけを指数バックオフとジッター付きで再試行し、SQLの誤りや認証エラーは即座に止めます。4番目が品質チェックで、ロードと同じトランザクションの中で検査し、不合格なら確定させません。最後が監視で、成功した最終時刻と件数を実行ログ表に残し、鮮度の遅れで気づける形にします。
全部を最初から入れる必要はありません。日次で数千行を1本流すだけなら、冪等な書き込みと実行ログだけで十分に運用できます。判断の目安は最後の章にまとめました。
検証環境の準備：PostgreSQL 18とPythonで動かすETLの最小構成
ツールに依存しない形で原則を確かめるため、抽出元も格納先もPostgreSQLにし、ドライバは psycopg 3系だけを使います。題材は注文テーブル orders を分析用の dw_orders へ毎時取り込む処理です。
ステージング・本表・ウォーターマーク・実行ログの4つの表を作る
格納先には4つの表を用意します。抽出した行をそのまま置く stg_orders、分析に使う本表 dw_orders、前回どこまで読んだかを持つ etl_watermark、実行のたびに1行残す etl_run_log です。ステージングに batch_id を持たせるのは、同じ実行をやり直したときに自分の行だけを消して入れ直すためです。
CREATE TABLE stg_orders (
order_id bigint NOT NULL,
customer_id bigint NOT NULL,
status text NOT NULL,
amount numeric(12,2),
updated_at timestamptz NOT NULL,
batch_id text NOT NULL
);
CREATE TABLE dw_orders (
order_id bigint PRIMARY KEY,
customer_id bigint NOT NULL,
status text NOT NULL,
amount numeric(12,2) NOT NULL,
updated_at timestamptz NOT NULL,
loaded_at timestamptz NOT NULL DEFAULT now(),
batch_id text NOT NULL
);
CREATE TABLE etl_watermark (
job_name text PRIMARY KEY,
last_value timestamptz NOT NULL
);
INSERT INTO etl_watermark VALUES ('orders', '1900-01-01');
CREATE TABLE etl_run_log (
run_id bigserial PRIMARY KEY,
job_name text NOT NULL,
batch_id text NOT NULL,
started_at timestamptz NOT NULL,
finished_at timestamptz,
status text NOT NULL,
rows_read integer,
rows_upserted integer,
error text
);stg_orders の amount だけ NULL を許しているのは意図的です。抽出元の汚れた値をいったん受け止め、品質チェックで弾くためで、ステージングで制約違反にすると原因の行が残らず調査に時間がかかります。
冪等な書き込みの実装：ステージング経由のUPSERTで再実行しても行が増えない形
Airflowの公式のベストプラクティスは、タスクは失敗時に再試行されるので毎回同じ結果を出すべきだとし、再実行時にINSERTを使わずUPSERTに置き換えるよう書いています。ツールが何であれ、この原則が全体の土台です。考え方の整理はデータパイプラインの冪等性と重複排除の設計に譲り、ここではSQLに落とします。
ON CONFLICT DO UPDATEに更新日時の比較を入れて古い行の上書きを防ぐ
本表への反映に必要なのは、次に示す1文のSQLです。PostgreSQL 18のINSERTでは、ON CONFLICT DO UPDATE に衝突判定の列を必ず指定し、提案された行は excluded で参照できます。WHERE を付けると更新する行を絞れます。
INSERT INTO dw_orders AS t
(order_id, customer_id, status, amount, updated_at, batch_id)
SELECT DISTINCT ON (order_id)
order_id, customer_id, status, amount, updated_at, batch_id
FROM stg_orders
WHERE batch_id = %(batch_id)s
ORDER BY order_id, updated_at DESC
ON CONFLICT (order_id) DO UPDATE
SET customer_id = excluded.customer_id,
status = excluded.status,
amount = excluded.amount,
updated_at = excluded.updated_at,
loaded_at = now(),
batch_id = excluded.batch_id
WHERE t.updated_at < excluded.updated_at;WHERE t.updated_at < excluded.updated_at の1行が効きます。同じバッチを2回流しても更新日時が同じなので何も起きず、遡り窓で古い版の行を読み直しても新しい値を巻き戻しません。%(batch_id)s は後述のPythonから渡すプレースホルダです。
同じ主キーが1バッチに2行あるとUPSERTが落ちる理由とDISTINCT ONの役割
上のSQLで DISTINCT ON を外すと、1時間のあいだに2回更新された注文があった時点で失敗します。PostgreSQLの公式ドキュメントは、ON CONFLICT DO UPDATE 付きのINSERTは1つの既存行を1コマンドで2回以上更新できず、その場合は cardinality violation のエラーになると明記しています。
開発環境では1行ずつしか更新が起きないため、このエラーは本番に出てから初めて見ることが多い種類です。ステージング側で主キーごとに最新の1行へ絞る処理は、UPSERTとセットで書く前提にしてください。BigQueryやSnowflakeのMERGEでも、ソース側に同じキーが重複すると失敗または非決定的になるので、考え方は共通です。
増分ロードの実装：ウォーターマーク表と遡り窓で遅れて届く行を拾う
毎回全件を読む設計は、数十万行を超えたあたりから抽出元のDBに負荷をかけ、実行時間も延びます。増分ロードは前回の到達点から先だけを読む方式で、到達点の決め方を誤ると黙って行が欠けます。
ウォーターマークを実行時刻ではなく読み取った行の最大更新日時で進める
よくある誤りは、ウォーターマークに「今回の実行時刻」を書くことです。抽出元のクロックとETLサーバのクロックがずれていると、その差の分だけ行が永久に読まれません。Airflowの公式ドキュメントがタスク内で now() を使わないよう書いているのも、同じ理由で再実行時に結果が変わるからです。
ウォーターマークは、実際に読み取った行の updated_at の最大値で進めます。0件のときは進めません。そしてこの更新は、本表へのUPSERTと同じトランザクションで確定させます。別トランザクションにすると、ロードが失敗したのにウォーターマークだけ進む状態が生まれます。
遡り窓30分で遅延コミットの取りこぼしを防ぐPython実装の全体像
もう1つの落とし穴が遅延コミットです。抽出元で updated_at が10時00分に付いた行でも、トランザクションのコミットが10時02分なら、10時01分に動いたETLからは見えません。次の回はウォーターマークの10時00分より後しか読まないため、この行は欠けます。対策は、ウォーターマークから一定時間さかのぼって読み直すことです。重複して読んだ行は前章のUPSERTが吸収します。
import random
import time
from datetime import timedelta
import psycopg
LOOKBACK = timedelta(minutes=30) # 遅延コミットを拾うための遡り幅
UPSERT_SQL = open("upsert_orders.sql", encoding="utf-8").read()
CHECK_SQL = open("check_orders.sql", encoding="utf-8").read()
def with_retry(fn, attempts=5, base=2.0, cap=60.0):
# 一時障害だけを再試行する。SQLの誤りなどは即座に例外を上げる
for n in range(1, attempts + 1):
try:
return fn()
except (psycopg.OperationalError, TimeoutError):
if n == attempts:
raise
# フルジッター：0秒から上限までのあいだでランダムに待つ
time.sleep(random.uniform(0, min(cap, base * 2 ** n)))
def run(src_dsn, dw_dsn, batch_id):
with psycopg.connect(dw_dsn) as dw:
wm = dw.execute(
"SELECT last_value FROM etl_watermark WHERE job_name = 'orders'"
).fetchone()[0]
since = wm - LOOKBACK
def extract():
with psycopg.connect(src_dsn, connect_timeout=10) as src:
return src.execute(
"SELECT order_id, customer_id, status, amount, updated_at "
"FROM orders WHERE updated_at >= %s",
(since,),
).fetchall()
rows = with_retry(extract)
with psycopg.connect(dw_dsn) as dw, dw.transaction():
# 同じbatch_idでやり直したときに自分の行だけを入れ直す
dw.execute("DELETE FROM stg_orders WHERE batch_id = %s", (batch_id,))
with dw.cursor().copy(
"COPY stg_orders (order_id, customer_id, status, amount, "
"updated_at, batch_id) FROM STDIN"
) as cp:
for r in rows:
cp.write_row((*r, batch_id))
check_batch(dw, batch_id) # 不合格なら例外でロールバック
upserted = dw.execute(UPSERT_SQL, {"batch_id": batch_id}).rowcount
if rows:
dw.execute(
"UPDATE etl_watermark SET last_value = %s WHERE job_name = 'orders'",
(max(r[4] for r in rows),),
)
return len(rows), upserted遡り幅は抽出元で最も長く走るトランザクションの時間より長く取ります。30分は、夜間バッチが抽出元を数分ロックする程度の業務システムを想定した値です。物理削除を拾う必要がある場合、この方式では検知できません。削除フラグに切り替えるか、ログを読む方式のCDC（Change Data Capture）を検討してください。
リトライの実装：指数バックオフとジッターで一時障害だけを再試行する
リトライは書き込みが冪等になってから足します。AWSのBuilders’ Libraryのリトライとバックオフの解説は、副作用のある処理は冪等でないかぎり再試行は安全ではないと書いています。前章までの形なら、何回流しても結果は同じです。
接続断とタイムアウトは再試行しSQLの誤りと認証エラーは即座に止める線引き
前章のコードで with_retry が捕まえるのは psycopg.OperationalError と TimeoutError の2つだけです。接続断、DBの再起動、一時的な接続数の上限は時間を置けば通る可能性があります。一方で列名の誤り（ProgrammingError）や制約違反（IntegrityError）、パスワードの誤りは、何回やっても同じ結果になります。
OperationalError には認証失敗も含まれる点に注意してください。認証エラーを5回再試行するとアカウントのロックを招くことがあるので、本番ではエラーコードを見て除外します。実務で最初に入れるべきなのは、この「再試行してはいけない例外」の一覧です。再試行の回数や間隔の調整はその後で構いません。
待ち時間を2の累乗で伸ばしつつ乱数で散らすフルジッターの書き方
待ち時間は base × 2 の n 乗で伸ばし、上限 cap で頭打ちにしたうえで、0からその値までの乱数を取ります。AWSの解説では、失敗した呼び出しが同じタイミングで一斉に再試行すると再び競合や過負荷を起こすため、ジッターで再試行の時刻を時間的に散らすという説明です。複数のETLが同じDBを読んでいる環境では、ジッターが無いと障害復旧の直後に全ジョブが同時に接続しにいきます。
同じ解説には、リトライはスタックの1か所で行うのが望ましいという指摘もあります。Pythonの中で5回、Airflowのタスク設定で4回と二重に持たせると、最悪で20回の接続になります。どちらで持つかを決め、もう一方は1回にしてください。
データ品質チェックの実装：件数・不正値・重複をロード直後に検査して止める
品質チェックはロードの後ではなく、確定させる前に置きます。前章のコードで check_batch を UPSERT の前に呼んでいるのはこのためで、例外を上げればトランザクションごとロールバックされ、本表もウォーターマークも動きません。
1本のSQLでバッチ単位の件数・NULL・範囲外の値・未知のステータスを数える
検査はステージングに対する集計SQL 1本で足ります。次の check_orders.sql は、バッチ内の件数と、金額の欠損や負の値、想定外のステータスを数えます。
SELECT count(*) AS rows_in_batch,
count(*) FILTER (WHERE amount IS NULL OR amount < 0) AS bad_amount,
count(*) FILTER (
WHERE status NOT IN ('placed', 'shipped', 'completed', 'returned')
) AS bad_status
FROM stg_orders
WHERE batch_id = %(batch_id)s;def check_batch(dw, batch_id):
rows, bad_amount, bad_status = dw.execute(
CHECK_SQL, {"batch_id": batch_id}
).fetchone()
if bad_amount or bad_status:
raise ValueError(
f"品質検査で不合格: batch={batch_id} 金額不正={bad_amount} 状態不正={bad_status}"
)この ValueError は with_retry の対象外なので、再試行されずに即座に失敗します。データの誤りは待っても直らないため、これが正しい振る舞いです。
0件や前回比の急減は止めずに警告に回す件数検査の閾値の分け方
件数の異常は扱いが分かれます。不正な値は1行でも止めますが、件数が0件や前回比50%未満に落ちたときは止めずに警告に回すのが実務的です。夜間や祝日に注文が減るのは正常で、ここで止めると毎週誤報が出ます。
止める検査と警告に回す検査を分ける考え方は、dbtやGreat Expectationsでも同じです。dbtはテストの severity を warn にすると下流を止めない設定にでき、Great Expectationsの検査とCheckpointの実装でも同じ使い分けができます。検査項目が数十を超えたら、SQLを手で書き続けるよりこうしたツールに移すほうが管理しやすくなります。
監視の実装：実行ログ表と鮮度・件数の閾値でETLの異常を検知する
ETLの障害で一番気づきにくいのは「エラーが出ずに止まっている」状態です。スケジューラ自体が落ちた、ジョブが無効化されたまま戻されていない、といった場合、失敗通知は1通も届きません。そこで、失敗を通知するのとは別に、成功が途絶えたことを検知します。
成功した最終時刻からの経過で止まったETLを拾う鮮度監視のSQL
run 関数の前後で etl_run_log に開始と終了、件数、エラー内容を書き込んでおけば、監視は次の1本で書けます。最後に成功してから2時間を超えたジョブだけを返すので、結果が1行でもあれば通知します。
SELECT job_name,
max(finished_at) FILTER (WHERE status = 'success') AS last_success,
now() - max(finished_at) FILTER (WHERE status = 'success') AS lag
FROM etl_run_log
GROUP BY job_name
HAVING coalesce(now() - max(finished_at) FILTER (WHERE status = 'success'),
interval '999 days') > interval '2 hours';毎時のジョブに2時間を閾値にするのは、1回の失敗はリトライに任せ、2回続けて落ちたときだけ人を呼ぶためです。日次ジョブなら26時間程度にします。
エラー文と件数を残す実行ログ表の項目設計と障害調査の時間短縮
実行ログに残すのは、batch_id、開始と終了の時刻、読み取り件数と反映件数、エラー文の5点です。障害の調査で最初に知りたいのは「いつから」「何件で」「何のエラーか」で、これがログ表にあればアプリケーションログを掘る前に当たりを付けられます。
読み取り件数と反映件数の差も見ておく値です。UPSERTの WHERE で更新を見送った行は反映件数に入らないので、遡り窓の分だけ差が出るのが正常です。差が普段より極端に大きいときは、抽出元で updated_at を更新しない一括修正が走った可能性があります。
Airflowとdbtへの載せ替え：再試行設定と増分モデルでの実装対応
ここまでの原則は、オーケストレーションや変換のツールに移しても形を変えて残ります。対応関係を押さえておけば、ツールの設定値を読むだけで設計の穴が分かります。以下で示すのは、再試行を retries 設定、増分ロードを incremental モデルに対応させる実装例です。
Airflow 3.4のDAGでの再試行設定と実行IDの対応
Airflow 3系ではTask SDKの dag と task でDAGを書きます。このDAGでは retries と指数バックオフを設定し、run_id を batch_id に渡す対応をコードに示します。Task SDK 1.4.0のAPIリファレンスでは、retry_exponential_backoff は真偽値ではなく再試行間隔の乗数（0で固定間隔）で、retry_delay の既定値は300秒です。
from datetime import datetime, timedelta
from airflow.sdk import dag, task
from orders_etl import run # 前章のrun関数
@dag(
schedule="@hourly",
start_date=datetime(2026, 11, 1),
catchup=False,
max_active_runs=1, # 同じジョブを並行させない
default_args={
"retries": 3,
"retry_delay": timedelta(minutes=2),
"retry_exponential_backoff": 2.0,
"max_retry_delay": timedelta(minutes=30),
"execution_timeout": timedelta(minutes=20),
},
)
def orders_etl():
@task
def load(run_id=None):
# 再試行でも同じrun_idになるので、ステージングの入れ直しが冪等になる
run(SRC_DSN, DW_DSN, batch_id=run_id)
load()
orders_etl()Airflowの側で retries を3にしたので、Python側の with_retry は attempts=1 にして二重化を避けます。データ誤りで再試行させたくない場合は、Airflowのタスク解説にある AirflowFailException を上げると、残りのリトライを無視して失敗にできます。
dbtのincrementalモデルとdata_testsで同じ増分と検査を書く対応表
変換をdbtに寄せる場合、dbtのincrementalモデルが本記事の増分ロードとUPSERTを1ファイルで表します。dbtの公式ドキュメントによると、unique_key を指定すると既存行を更新し、指定しないと多くのアダプタで追記だけになります。
{{ config(
materialized='incremental',
unique_key='order_id',
on_schema_change='fail'
) }}
select order_id, customer_id, status, amount, updated_at
from {{ source('app', 'orders') }}
{% if is_incremental() %}
where updated_at >= (
select coalesce(max(updated_at), '1900-01-01') from {{ this }}
) - interval '30 minutes'
{% endif %}models:
- name: fct_orders
columns:
- name: order_id
data_tests:
- unique
- not_null
- name: status
data_tests:
- accepted_values:
arguments:
values: ['placed', 'shipped', 'completed', 'returned']
sources:
- name: app
tables:
- name: orders
config:
loaded_at_field: updated_at
freshness:
warn_after: {count: 2, period: hour}
error_after: {count: 6, period: hour}dbtのデータテストは unique・not_null・accepted_values・relationships の4種が標準で、テスト引数を arguments の下に書く形は v1.10.5 以降です。freshnessの設定は v1.9 から config の下に移り、loaded_at_field も v1.10 から同じ場所に置く設定です。dbt buildでは上流のテストが失敗すると下流のモデルがスキップされるため、品質チェックで確定を止める動きもそのまま再現できます。取り込みから配信までを dlt と dbt で組む手順は、データパイプライン構築の手順で扱っています。
ベストプラクティスを全部入れない判断：小規模ETLで省いてよい項目
5つの原則をすべて実装すると、コードは最初の数倍になります。規模に対して過剰な作りは、保守する人がいなくなった時点で負債に変わります。
日次数千行のETL1本なら全件洗い替えと実行ログだけで足りる条件
次の3つが揃うなら、増分ロードとリトライは入れなくて構いません。対象が日次で数千行から数万行に収まる、失敗しても翌朝の手動再実行で業務に支障が出ない、抽出元に別の利用者がいない、の3つです。この規模なら、毎回 TRUNCATE してから全件を入れ直すほうが単純で、遅延コミットも削除も気にせずに済みます。
ただし、洗い替えでも全件の入れ替えは1トランザクションで行い、実行ログだけは必ず残してください。この2つを省くと、途中で落ちたときに表が空のまま朝を迎え、しかも誰も気づきません。省いてよいのは増分とリトライで、冪等性と監視は省けない、というのがこの記事の結論です。
自前実装をやめてマネージドのETLツールに寄せるべき3つの兆候
逆に、抽出元が10を超えた、SaaSのAPIから取り込む系統が増えた、スキーマ変更への追随に毎月人手を取られている、のいずれかに当てはまったら、自前のコードを書き足すより製品に寄せる時期です。コネクタの保守とリトライはツール側が持つので、自社は品質チェックと監視の設計に集中できます。製品ごとの課金単位と料金はETLツール比較7製品の料金と課金単位で整理しました。
どこまでを自前で持ち、どこからを製品に任せるかは、データ量と社内の運用体制の両方で決まります。既存ETLの棚卸しから冪等化や監視の後付けまでを含めて相談したい場合、当社のデータ分析基盤構築・MLOps構築支援で設計段階から支援しています。
よくある質問
ETLのベストプラクティスを実装に移す段階でよく挙がる質問を5つ取り上げます。
ETLの冪等性とは何ですか？
同じ入力で何回実行しても、格納先の状態が同じになる性質です。ETLでは、失敗後の再実行や、過去期間のやり直し（バックフィル）が日常的に起きるため、この性質が無いと再実行のたびに行が重複します。主キーを決めてUPSERTで書き込み、ステージングを実行単位の batch_id で入れ直す形にすれば、多くのETLは冪等になります。
増分ロードと全件ロードはどちらを選ぶべきですか？
対象が日次数万行程度までで、処理時間と抽出元の負荷に余裕があるなら全件ロードのほうが単純で、削除の反映も自然にできます。数十万行を超える、または時間単位以下の鮮度が必要な場合は増分ロードにします。増分にするなら、ウォーターマークの進め方と遡り窓をセットで設計してください。
ETLのリトライ回数は何回が目安ですか？
回数より先に、どの例外を再試行するかを決めてください。一時障害だけを対象にするなら、3〜5回で指数バックオフの上限を数十分に置く設定が一般的です。リトライはPythonとスケジューラのどちらか1か所に持たせ、二重にしないことが回数の調整より効きます。
ETLのデータ品質チェックは何から始めればよいですか？
主キーの一意性とNULL、それに業務上ありえない値（負の金額、未知のステータス）の3つから始めます。いずれも集計SQL 1本で書け、不合格なら確定させないという単純なルールで運用できます。件数の増減は誤報が多いので、止めずに警告として扱うところから入るのが無難です。
ETLの監視では何を見ればよいですか？
最後に成功した時刻からの経過時間を最優先にします。失敗通知だけでは、スケジューラが止まったときに気づけないためです。次に、実行ごとの読み取り件数と反映件数、所要時間を記録し、普段との差で異常に気づける状態にします。
関連記事
- データパイプラインとは？種類・構成要素と冪等性から考える設計判断を解説【2026年版】：本記事の原則を概念と構成の側から整理した親記事です。
- ETLとは？仕組み・ELTとの違い・ツール選定から導入判断まで解説：ETLそのものの定義と導入判断を扱っています。
- データパイプライン構築の手順：dltで取込・dbtで変換・Parquetで配信する最小構成とコストの増え方【2026年版】：ツールを使って取り込みから配信までを組む手順です。
- Great Expectationsとは？GX Core 1.xの部品構成・実装手順と採用判断を解説【2026年版】：品質検査をツールに移すときの選択肢です。
- AWS GlueでETLジョブを動かす手順｜PySparkスクリプトとDPU課金の設計：AWS上のマネージドETLで同じ処理を動かす場合の手順です。
