data-pipeline-builder

数据管道构建器

设计并实现生产级 ETL/ELT 数据管道：接收数据源、目标位置和转换需求，然后生成完整的管道规范，以及运行所需的全部实现文件。

目录

• references/project-structure.md -- 输出文件布局、架构模式选择、组件选择。
• references/python-patterns.md -- Python 代码规范，以及提取器/转换器/加载器/重试的基础模式。
• references/quality-checks.md -- 可组合的数据质量检查框架和内置检查。
• references/orchestration-config.md -- Airflow DAG、管道配置 YAML，以及监控/告警模式。
• references/spec-template.md -- data-pipeline-spec.md 输出模板。

工作流程

1. 收集需求。如果用户已提供明确需求，则继续设计；否则，提出有针对性的问题：数据源（数据库、API、文件、流）；目标位置（数仓、数据湖、数据库）；转换（连接、聚合、筛选、业务规则）；数据新鲜度要求（实时、每小时、每天）；技术偏好（Airflow、dbt、Spark、云服务商）；数据质量和合规要求。

2. 分析并设计。梳理每个数据源（连接类型、身份验证、架构、数据量、CDC 可用性、速率限制）。定义目标位置（平台、架构设计、分区、聚簇、访问模式）。规划转换（字段映射、业务逻辑、类型转换、连接、聚合、去重、SCD 处理、派生字段）。确定非功能性需求（新鲜度 SLA、处理时段、故障容忍度、保留期限、合规性）。根据 references/project-structure.md 选择架构模式和组件。

3. 在生成代码前展示设计方案。与用户确认架构、数据源、目标位置、运行计划、关键转换和质量门禁，并在获得批准后继续。

4. 生成实现。按照 references/project-structure.md 中的布局生成所有文件，并根据具体管道定制，不得包含需要手动编辑的占位代码：

   • 为每个数据源生成一个继承自 BaseExtractor 的具体提取器（参见 references/python-patterns.md）。
   • 为每项转换生成一个具体的转换器类或 SQL 文件。
   • 为每个目标位置生成一个继承自 BaseLoader 的具体加载器。
   • 生成连接好所有任务依赖项的 Airflow DAG 和管道配置 YAML（参见 references/orchestration-config.md）。
   • 生成针对数据定制的质量检查和包含适当告警阈值的监控配置（参见 references/quality-checks.md 和 references/orchestration-config.md）。
   • 为所有自定义业务逻辑生成测试。

5. 最后生成规范。使用 references/spec-template.md 生成 data-pipeline-spec.md，引用所有实现文件，并纳入过程中作出的设计决策。

操作规则

• 设计时确保幂等性——让每个步骤都能安全地重新运行。
• 为增量管道加入水位线/检查点跟踪。
• 为处理失败的记录加入死信处理。
• 加入架构演进处理机制——数据源的架构会发生变化。
• 切勿硬编码凭据——使用环境变量或密钥管理器。
• 切勿跳过质量检查——它们是防范错误数据的第一道防线。
• 对可用 SQL 表达的转换优先使用 SQL；对于无法自然映射为 SQL 的复杂逻辑，则使用 Python。
• 在规范中加入回填策略和涵盖常见故障场景的运维手册。
• 全程使用结构化日志，并在每个转换步骤跟踪数据沿袭。