etl-patterns

ETL 模式

用于生产级提取-转换-加载模式的编排器。

技能路由

需求	技能	内容
可靠性模式	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 转换