etl-patterns

ETL パターン

本番環境向けの Extract-Transform-Load パターンを実現するオーケストレーターです。

スキルの振り分け

必要な内容	スキル	内容
信頼性パターン	etl-core-patterns	冪等性、チェックポイント、エラー処理、チャンク処理、リトライ、ロギング
ロード戦略	etl-incremental-patterns	バックフィル、タイムスタンプベース、CDC、パイプラインのオーケストレーション

パターン選択ガイド

信頼性のニーズ別

必要な内容	パターン	スキル
繰り返し実行	冪等性	etl-core-patterns
障害後の再開	チェックポイント	etl-core-patterns
不正なレコードへの対応	エラー処理＋DLQ	etl-core-patterns
メモリ管理	チャンク処理	etl-core-patterns
ネットワークの耐障害性	バックオフ付きリトライ	etl-core-patterns
オブザーバビリティ	構造化ログ	etl-core-patterns

ロード戦略別

シナリオ	パターン	スキル
小規模テーブル（<100K）	全件リフレッシュ	etl-incremental-patterns
大規模テーブル	タイムスタンプによる増分処理	etl-incremental-patterns
リアルタイム同期	CDC イベント	etl-incremental-patterns
履歴データの移行	並列バックフィル	etl-incremental-patterns
ダウンタイムなしのリフレッシュ	スワップパターン	etl-incremental-patterns
複数ステップのパイプライン	パイプラインのオーケストレーション	etl-incremental-patterns

クイックリファレンス

冪等性の選択肢

# Small datasets: Delete-then-insert
# Large datasets: UPSERT on conflict
# Change detection: Row hash comparison

ロード戦略の決定

Is table < 100K rows?
  → Full refresh

Has reliable timestamp column?
  → Timestamp incremental

Source supports CDC?
  → CDC event processing

Need zero downtime?
  → Swap pattern (temp table → rename)

One-time historical load?
  → Parallel backfill with date ranges

一般的なパイプライン構造

# 1. Setup
checkpoint = Checkpoint('.etl_checkpoint.json')
processor = ETLProcessor()

# 2. Extract (with incremental)
df = incremental_by_timestamp(source_table, 'updated_at')

# 3. Transform (with error handling)
transformed = processor.process_batch(df.to_dict('records'))

# 4. Load (with idempotency)
upsert_records(pd.DataFrame(transformed))

# 5. Checkpoint
checkpoint.set_last_processed('sync', df['updated_at'].max())

# 6. Handle failures
processor.save_failures('failures/')

関連スキル

• data-validation - ETL 中にデータ品質を検証します
• data-quality - データ品質の指標を監視します
• pandas-coder - DataFrame の変換