ETL パイプライン
PostgreSQL（デフォルトストア）へのバッチ取り込みを Python（uv, ruff, pytest）で行います。ローカル変換には Polars 2.x または DuckDB 2.0 を使用します。
手順
- グレイン（粒度）を定義する: 1 行が何を意味し、どの自然キーで識別されるかを明確にします。キーがなければ冪等ロードはできません。キー（またはコンテンツハッシュ）を最初に定義してください。
- ロードパターン（後述）を選択します。インクリメンタル戦略を選択します。オーケストレーターは最後に選びます。
- 任意のウィンドウをリランしても同じ最終状態になるように処理を記述します。テストで 2 回実行し、行数とチェックサムが不変であることをアサートして証明します。
- 本番実行前に オブザーバビリティチェック（後述）を追加します。
- 最後のレビューチェックリストで確認します。
冪等ロードパターン
- 行に安定した自然キーがあり、値が変わる可能性がある場合:
INSERT ... ON CONFLICT (key) DO UPDATE（Postgres）またはMERGEを使用します。WHERE target.hash IS DISTINCT FROM excluded.hashを追加して、変更のない行が更新されないようにします。 - データがパーティション全体（日付、バッチID）として到着し、単位として書き換えられる場合: そのパーティションの DELETE-INSERT を 1 つのトランザクションで行います。DELETE と INSERT が別々にコミットされると、クラッシュ時に穴が残ります。
- テーブル全体を再構築する場合: ステージングテーブルにロードし、検証後、1 つのトランザクションでリネーム（スワップ）します。リーダーが中途半端にロードされたテーブルを見ることはありません。
- キーのない追加専用ファクトの場合: 決定論的な
batch_id/ ソースファイルハッシュを追加し、ロード時に既にロード済みのバッチ ID を拒否または置換するようにします。 - 決して
INSERTを無条件に行い、後から重複除去してはいけません。重複は重複除去が実行される前にダッシュボードに到達します。 COPYを使用して一時/ステージングテーブルにロードし、その後 1 回のセットベースの upsert を実行します。Python からの行単位の upsert は 10～100 倍遅くなります。
インクリメンタルロード
- ウォーターマークには、ソースが単調増加かつインデックス付けされていることが保証されているカラム（updated_at, シーケンスID）を使用します。ロードと同じトランザクションで書き込まれる状態テーブルに格納し、決してサイドのファイルには保存しません。
watermark - lookback（例: 1～24 時間）で読み取り、> watermarkでは読み取りません。ソースのコミットはタイムスタンプ順に並んでいるとは限りません。ルックバックと upsert で遅延行を吸収します。ルックバックのサイズは、推測ではなく、測定された遅延に基づいて決定します。- ウォーターマークカラムが信頼できない場合（更新時にバンプされない、タイムゾーンが混在する、ソフトデリートが欠落するなど）：定期的にフルリコンサイル（キーセットの差分、またはパーティションごとのカウント＋チェックサム）を実行し、ウォーターマークを信頼しないでください。
- ソースでのハードデリートはウォーターマークでは検出できません。明示的に判断します：キー差分でリコンサイルするか、受け入れて文書化します。
- CDC（Postgres 論理レプリケーション、Debezium スタイルのログリーダー）：削除、サブミリ秒のレイテンシ、またはソースをタイムスタンプでクエリできない場合にのみ使用します。スロットの追加、保持リスク（アイドルスロットがソースディスクを満たす）、順序/スナップショット処理が発生します。それ以外の場合はウォーターマークを使用します。
バックフィル
- すべてのジョブは
--startと--end（半開区間[start, end)、UTC）を受け取り、すべてをこれらから導出します。変換内でnow()を使用しないでください。 - バックフィルとは、通常のジョブを過去のウィンドウに対して実行することです。異なるコードが必要な場合、通常のジョブは冪等ではありません。修正してください。
- パーティションごとにチャンク化し、各チャンクを独立して実行し、チャンクごとのステータスを記録します。これにより、チャンク 40 で失敗した場合、40 から再開できます。
- 同じ upsert/パーティション置換を使用して同じターゲットにバックフィルします。二重計上を確認します：ウィンドウの前後で
count(*)とcount(distinct key)を確認します。 - ソース API/DB に対するスロットルを実施します。バックフィルは自己誘発的な負荷スパイクです。
リトライ、タイムアウト、「エグザクトリワンス」
- ネットワーク越しのエグザクトリワンス配信は存在しません。アットリースワンス配信 ＋ 冪等な書き込みを構築します。結果は実質的にワンス（effectively-once）になります。
- 冪等なステップのみをリトライし、指数バックオフとジッター、試行回数の上限、すべてのネットワークコールにハードタイムアウト（接続タイムアウトと読み取りタイムアウトは個別に設定）を設けます。429/5xx/タイムアウトでリトライし、4xx ではリトライしません。
Retry-Afterを尊重します。グローバルレート制限なしのリトライは障害を増幅させます。- メール送信、課金、Webhook 投稿を行うステップは冪等ではありません：呼び出し前に保存された冪等キーでガードします。
- 最終失敗時は非ゼロで終了しアラートを送信します。スケジュールを正常に保つために例外を飲み込まないでください。
スキーマドリフト
- ランディングレイヤーは寛容にします（生ペイロードを JSON/テキスト＋ロードメタデータとして保存）。型付きレイヤーは厳格にし、型や必須フィールド違反は大声で失敗させます。
- 新しいカラム：追加は許可しますが、アラートを出します。削除、名前変更、型変更されたカラム：実行を失敗させ、差分を表示します。NULL に強制変換しないでください。
- 明示的なスキーマ（polars スキーマ、pydantic、または DuckDB の
DESCRIBE）で境界で一度だけ検証し、変換全体に散在させないでください。
スクレイプからウェアハウスへ
- 2 段階方式。ステージ 1 では生のレスポンスを変更せず不変にフェッチして保存します（オブジェクトパスはソース、フェッチ時刻、コンテンツハッシュでキー指定）。ステージ 2 で生データをテーブルにパースします。
- パーサーのバグは、保存された生データを再パースすることで修正され、再フェッチ、禁止リスク、履歴の喪失はありません。
- URL、フェッチタイムスタンプ、HTTP ステータス、パーサーバージョンをすべてのパース済み行に記録します。パースは生バイトとパーサーバージョンの純粋関数です。
- フェッチ失敗や空/ブロックされたページ（キャプチャ、ログインウォールのある 200）は、ステータスコードだけでなく、コンテンツチェックによって検出します。
オブザーバビリティ（最小限）
- ステップごとにログ：読み取った行数、書き込んだ行数、拒否した行数、処理時間、ウィンドウ開始/終了、実行 ID。構造化（JSON）行で出力します。
- 不変条件をアサート：出力行数と入力行数を許容範囲内で比較。通常は空でないウィンドウで 0 行の場合は実行を失敗させます。0 行の成功を黙認することは、最も一般的な未検出障害です。
- ハートビート：成功時と開始時に外部ヘルスチェックに ping を送信します。ping の欠落があった場合にアラートを送信します（cron 未実行、ホストダウンを検出）。失敗時も失敗エンドポイントに ping を送信します。
- フレッシュネスアラート（ロードされた最新のタイムスタンプがしきい値を超えて古い場合）を、ジョブエラーだけでなく設定します。
- 実行ログテーブル（実行 ID、ウィンドウ、ステータス、カウント）を保持します。これはバックフィル元帳としても機能します。
オーケストレーターの選び方
- cron（または systemd タイマー）＋ 1 つのスクリプト：少数の独立したジョブ、ジョブ間の依存関係なし、ログを読めるチーム。デフォルト。上記のハートビートと、重複実行を防ぐロック（
flock）が必要です。 - n8n：低ボリューム、SaaS API 間のグルー、Webhook トリガー、非開発者がフローを編集。大規模バッチ、重い変換、コードレビューには不向き。重いロジックは、n8n が呼び出す Python スクリプトに任せてください。
- Airflow 3.x：多数の DAG、ジョブ間の依存関係、ファーストクラスのバックフィル、リトライと SLA、監査 UI。実際の運用コスト（スケジューラー、メタデータ DB、ワーカー）がかかります。少数のジョブには価値がありません。Airflow 2 から 3 への移行は破壊的変更を伴います。スニペットをコピーする前にバージョンを確認してください。dbt on Airflow には Astronomer Cosmos を使用します。
- dlt：スキーマ推論、インクリメンタル状態、マージ/リプレイス処理を自動化したい API またはデータベースの取り込みに使用します。手動でページネーション抽出＋ウォーターマーク＋upsert コードを書く代わりに使用します。上記のルールをその設定にも適用してください。
- コンテナ：シークレットは Compose secrets として
/run/secrets/<name>から読み取り、環境変数や env_file は使用しません。公開ポートは127.0.0.1にバインドするか、内部ネットワーク上に配置します。
ベンダースキル
dbt や Airflow の詳細（モデル、テスト、DAG 作成、オペレーター、Cosmos 設定）が必要なタスクでは、即興で実装せずに、このリポジトリの de-vendors スキルを通じて取り込まれた公式ベンダースキルを使用し、references/landscape-2026-11.md のバージョンは最新情報の確認にのみ使用します。
レビューチェックリスト
- キーとグレインが明確に定義され、同じウィンドウのリランで同一の状態になる（テスト済み）。
- 削除と挿入が 1 つのトランザクションで行われ、ウォーターマークがロードと共にコミットされる。
- ルックバックが存在し、遅延データと削除に対する明示的な回答がある。
- タイムアウト、バックオフ付きの制限付きリトライ、非冪等ステップがガードされている。
- 生のランディングが不変で、パースが分離可能かつ再実行可能である。
- ステップごとに行数がログに記録され、0 行アラートとフレッシュネスアラートが存在し、ハートビートが設定されている。
- スキーマドリフトが型付きレイヤーで大声で失敗する。
- オーケストレーターの選択が、習慣ではなくジョブ数と依存関係によって正当化されている。
具体的な SQL と Python のスケルトンは references/patterns.md にあります。実際の使用からの教訓は references/gotchas.md にあります。