name: data-pipeline-architect description: 设计ETL/ELT数据管道，包含正确的提取、转换和加载模式，以及编排、错误处理和数据质量验证。 license: MIT
数据管道架构师
该技能提供设计稳健、可扩展数据管道的指导，可靠地将数据从源移动到目的地。
核心能力
- ETL vs ELT: 传统提取-转换-加载与现代提取-加载-转换模式
- 编排: Airflow、Dagster、Prefect、dbt用于工作流管理
- 数据质量: 验证、监控、血缘跟踪
- 可扩展性: 批处理 vs 流处理、分区、并行化
管道设计过程
1. 需求分析
开始管道设计时，收集：
- 源系统和数据格式（API、数据库、文件、流）
- 目标目的地（数据仓库、数据湖、湖仓一体）
- 新鲜度要求（实时、每小时、每天）
- 数据量和速度估计
- 质量和合规要求
2. 架构选择
批处理管道 - 用于周期性批量处理：
- 计划驱动（每小时、每天、每周）
- 更高的延迟容忍度
- 更简单的错误恢复（重新运行整个批次）
- 工具：Airflow、dbt、Spark
流处理管道 - 用于实时要求：
- 事件驱动处理
- 亚秒到分钟延迟
- 复杂状态管理
- 工具：Kafka、Flink、Spark Streaming
混合方法 - Lambda或Kappa架构：
- 批处理层用于完整性
- 速度层用于低延迟
- 服务层用于查询
3. ETL vs ELT 决策
ETL（加载前转换）:
- 当目标计算能力有限时
- 当转换显著减少数据体积时
- 当敏感数据必须在落地前屏蔽时
- 传统数据仓库模式
ELT（加载后转换）:
- 现代云仓库计算成本低
- 当需要保留原始数据时
- 当转换频繁变化时
- dbt风格在仓库中的转换
4. 管道组件
提取层:
- 完全提取 vs 增量（CDC、基于时间戳）
- API分页和速率限制
- 连接池和重试逻辑
- 模式检测和漂移处理
转换层:
- 数据清洗和标准化
- 业务逻辑应用
- 聚合和反规范化
- 类型转换和空值处理
加载层:
- 更新策略（合并、删除+插入）
- 分区方案（时间、哈希、范围）
- 索引管理
- 事务边界
5. 错误处理模式
┌─────────────────────────────────────────────────────────┐
│ 管道执行 │
├─────────────────────────────────────────────────────────┤
│ ┌─────────┐ ┌───────────┐ ┌──────────┐ │
│ │ 提取 │───▶│ 转换 │───▶│ 加载 │ │
│ └────┬────┘ └─────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────┐ ┌───────────┐ ┌──────────┐ │
│ │ 重试 │ │ 死信队列 │ │ 回滚 │ │
│ │ 带退避 │ │ │ │ 检查点 │ │
│ └─────────┘ └───────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────┘
- 带退避的重试: 临时故障（网络、速率限制）
- 死信队列: 无法处理的毒消息
- 检查点: 从最后一个成功点恢复
- 幂等性: 安全重新运行而无重复
6. 数据质量框架
在每个阶段实施检查：
| 阶段 | 检查类型 | 示例 |
|---|---|---|
| 提取 | 完整性 | 行数与源匹配 |
| 提取 | 新鲜度 | 数据时间戳在SLA内 |
| 转换 | 有效性 | 值在预期范围内 |
| 转换 | 唯一性 | 主键唯一 |
| 加载 | 对账 | 目标匹配源总计 |
| 加载 | 完整性 | 外键有效 |
7. 监控和可观测性
要跟踪的基本指标：
- 管道持续时间和趋势
- 每个阶段的行数
- 错误率和类型
- 数据新鲜度（自上次成功运行以来的时间）
- 资源利用率
警报：
- SLA违规（数据不新鲜）
- 异常行数（与基线±20%）
- 源中的模式变化
- 重复失败
常见模式
缓慢变化维度（SCD）
- 类型1: 覆盖（无历史）
- 类型2: 添加带有效日期的行
- 类型3: 先前值列
- 类型4: 历史表
增量处理
-- 基于时间戳的增量
SELECT * FROM source
WHERE updated_at > {{ last_run_timestamp }}
-- CDC-based (Change Data Capture)
-- 从事务日志捕获插入、更新、删除
幂等加载
-- 删除 + 插入模式
DELETE FROM target WHERE date_partition = '2024-01-15';
INSERT INTO target SELECT * FROM staging WHERE date_partition = '2024-01-15';
-- 合并/更新插入模式
MERGE INTO target t
USING staging s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...
参考
references/orchestration-patterns.md- Airflow、Dagster、Prefect模式references/data-quality-checks.md- 验证框架和规则references/pipeline-templates.md- 常见管道架构