データ取り込みパイプラインの構築
使用する場合
- データベース、API、ファイル、SaaSからウェアハウス/レイクへの抽出。
- 差分抽出、ウォーターマーク、またはCDCの設計。
- APIのページネーション、レート制限、再試行の処理。
- マネージドEL（Fivetran/Airbyte）とカスタムコードの決定。
- すでに取り込まれたデータの変換には使用しないでください（dbt/Sparkスキルを使用）。
ワークフロー
- [ ] 抽出モードを決定する：フルスナップショット vs 差分 vs CDC
- [ ] 信頼性の高いハイウォーターマークを選ぶ（updated_at、LSN/binlog、シーケンス）
- [ ] 生データを不変に保存（追記）、後続の変換は下流で行う
- [ ] ロードをべき等にする（キーによるupsert/パーティション上書き）
- [ ] ページネーション、レート制限、再試行、遅延データを処理する
- モードを選択する。 フルリロード（小規模/ディメンションテーブル）、ハイウォーターマークによる差分（ほとんどのファクトテーブル）、またはCDC（削除とすべての変更が必要な高ボリュームOLTP）。
- 信頼性の高いウォーターマークを選ぶ。
updated_atはソースが常にそれを更新する場合にのみ機能します。それ以外の場合は、DBログ位置（LSN/binlog/SCN）または単調増加シーケンスを使用します。最後のウォーターマークを保存し、そこから再開します。 - 生データを不変に保存する。 生の抽出データをロードメタデータとともに追記（Bronze）します。変換は下流で行い、ソースから再取得せずに再実行できるようにします。
- べき等ロード。 自然キーでupsertするかパーティションを上書きし、再試行やオーバーラップで重複が発生しないようにします（
writing-idempotent-transformationsを参照）。 - 堅牢に処理する ページネーション、レート制限、遅延データに対して。
パターン
遅延データ対策のオーバーラップ付き差分抽出：
-- 最後のウォーターマークを過ぎた小さなオーバーラップウィンドウをプルして遅延更新をキャッチし、
-- キーでupsertすることでオーバーラップによる重複を防ぐ。
SELECT * FROM source.orders
WHERE updated_at >= :last_watermark - INTERVAL '3 days';
CDC — データベースログ（Debezium/ネイティブ）を読み取り、挿入、更新、削除（ウォーターマークベースの抽出では見逃される）をキャプチャします。変更のコミット位置を使用して順序付けし、MERGEで変更を適用します。
APIページネーション + レート制限 — カーソル/次ページトークンに従います。429/5xxに対して指数バックオフとジッターでバックオフし再試行します。進捗をチェックポイントとして保存し、障害時に途中から再開できるようにします。
マネージド vs カスタム — 標準コネクタにはFivetran/Airbyteを使用します（スキーマ変更、差分状態、再試行を処理します）。カスタムコードは、特殊なソースや厳格な制御/コスト要件がある場合のみ作成します。
よくある落とし穴
- ソースが確実に更新しない
updated_atウォーターマーク — 行を静かに見逃します。検証するか、ログベースのCDCに切り替えます。 - オーバーラップウィンドウがない — 実行間に遅れて到着した更新が失われます。
- ウォーターマークベースの抽出で削除を期待する — 削除を検出できません。CDCまたは定期的なフルリコンシリエーションを使用します。
- 抽出中に変換を行う — 再実行が不可能になります。最初に生データを保存します。
- レート制限/ページネーションのエッジケースを無視する — 完全に見えるが実際は部分的なプルになります。チェックポイントを設定し、カウントを検証します。
- 非べき等ロード — 再試行とオーバーラップウィンドウにより行が重複します。