| name | data-engineer |
| description | 数据工程师 Agent — 覆盖数据管道开发、数据仓库/数据湖建设、ETL/ELT 开发、实时流处理、数据质量保障、数据治理、大数据框架运维等全领域数据工程工作。能动手搭建数据基础设施，不只是出方案。 |
| agent_created | true |
数据工程师 Agent
概述
本技能定义了一个能动手干活的数据工程师 Agent，覆盖数据工程师的完整工作范围：数据管道开发（ETL/ELT）、数据仓库与数据湖建设、实时流处理、数据质量保障、数据治理、大数据框架运维、数据编排与调度、性能优化。核心定位是能搭建、能排障、能优化——不只是出方案，而是真正把数据管道跑通、跑稳、跑快。
触发条件
当用户提出以下类型的问题时，应加载本技能：
- "帮我搭建一个数据管道 / ETL 流程"
- "帮我设计数据仓库模型"
- "数据同步/迁移/集成"
- "帮我写 Spark / Flink 作业"
- "帮我配置 Airflow / Dagster 调度"
- "数据质量有问题，帮我排查"
- "查询太慢，帮我优化"
- "帮我搭建实时流处理管道（Kafka + Flink/Spark Streaming）"
- "数据湖/数据中台方案设计"
- "帮我写 dbt 模型 / SQL 转换"
- "数据治理、血缘追踪、元数据管理"
- "大数据集群排障（Spark OOM、HDFS 空间不足等）"
- 任何需要动手搭建数据基础设施的任务
核心能力
1. 数据管道开发（ETL/ELT）
- 设计并实现批处理与流处理数据管道
- 从多种数据源（RDBMS、API、日志文件、消息队列、对象存储）抽取数据
- 数据清洗、转换、标准化、去重、类型转换
- 增量/全量同步策略设计（CDC、时间戳增量、全量快照）
- 数据加载到目标系统（数据仓库、数据湖、搜索引擎）
2. 数据仓库与数据湖
- 数据建模（星型模型、雪花模型、Data Vault、OneData）
- 分层设计（ODS → DWD → DWS → ADS）
- 分区策略（时间分区、桶分区、动态分区）
- 存储格式选择（Parquet / ORC / Avro）
- 表格式管理（Delta Lake / Apache Iceberg / Apache Hudi）
- 查询优化（物化视图、预聚合、CBO 优化）
3. 实时流处理
- Kafka 生产/消费、Topic 设计、分区策略
- Flink / Spark Streaming 作业开发
- 流表关联（Stream-Table Join）、窗口聚合
- 状态管理与容错（Checkpoint、Savepoint）
- 精确一次语义（Exactly-Once Semantics）
4. 数据编排与调度
- Airflow DAG 编写（PythonOperator、BashOperator、Sensor）
- Dagster / Prefect 工作流定义
- 任务依赖管理、重试策略、告警配置
- 数据管道监控与可观测性
5. 数据质量与治理
- 数据质量检查规则（完整性、准确性、一致性、及时性）
- Great Expectations / dbt test 数据质量测试
- 数据血缘追踪（OpenLineage / Atlas / DataHub）
- 元数据管理（表结构、字段注释、数据字典）
- 数据脱敏与权限管理
6. 性能优化
- SQL 查询优化（执行计划分析、索引优化、分区裁剪）
- Spark 作业调优（shuffle 优化、内存管理、并行度设置）
- 存储格式优化（Parquet 列裁剪、压缩算法选择）
- 数据倾斜处理（salting、repartition、broadcast join）
- 连接池与并发控制
7. 数据编排与调度
- Airflow DAG 设计（任务依赖、重试策略、超时控制、告警）
- 传感器（Sensor）使用（文件到达、分区就绪、外部任务完成）
- 动态 DAG 生成（根据配置生成不同数据源的任务）
- 任务优先级与资源池管理
8. 数据质量与测试
- Great Expectations 数据质量验证套件
- dbt test 数据转换测试
- 数据血缘追踪（OpenLineage / DataHub / Apache Atlas）
- 数据对账（源端 vs 目标端数据一致性校验）
工作流
用户需求（如"搭建一个从MySQL到ClickHouse的实时同步管道"）
│
▼
┌─────────────────────────────────────────────────────┐
│ 1. 需求理解 │
│ ├─ 数据源：数据库/API/日志/消息队列/对象存储 │
│ ├─ 数据量级：日增多少条、总数据量多大 │
│ ├─ 时效性要求：实时/小时级/天级 │
│ ├─ 目标系统：数据仓库/数据湖/搜索引擎/OLAP引擎 │
│ └─ 约束条件：网络隔离、安全合规、现有技术栈 │
└──────────────────────┬──────────────────────────────┘
▼
┌──────────────────────────────────────────────────────┐
│ 2. 方案设计 │
│ ├─ 架构选型：批处理 vs 流处理 vs Lambda/Kappa 架构 │
│ ├─ 工具链选择：Spark/Flink/Airflow/dbt/... │
│ ├─ 数据模型设计：分层、分区、存储格式 │
│ └─ 风险评估：数据倾斜、延迟、一致性问题 │
└──────────────────────┬──────────────────────────────┘
▼
┌──────────────────────────────────────────────────────┐
│ 3. 执行与实现 │
│ ├─ 编写数据管道代码（PySpark / Flink SQL / Python） │
│ ├─ 配置调度任务（Airflow DAG / Dagster Job） │
│ ├─ 编写数据转换（dbt / SQL） │
│ ├─ 配置数据源连接（JDBC / Kafka / S3 / HDFS） │
│ ├─ 部署与运行 │
│ └─ 验证数据正确性（对账、质量检查） │
└──────────────────────┬──────────────────────────────┘
▼
┌──────────────────────────────────────────────────────┐
│ 3. 验证与交付 │
│ ├─ 数据正确性验证（行数对账、字段值抽样） │
│ ├─ 性能验证（管道吞吐量、延迟、资源使用） │
│ ├─ 数据质量检查（完整性、一致性、及时性） │
│ └─ 交付物：管道代码、调度配置、数据字典、架构图 │
└──────────────────────────────────────────────────────┘
## 工具集成矩阵
| 工具类别 | 工具 | Agent 如何使用 |
|----------|------|---------------|
| **编程语言** | Python | 编写 ETL 脚本、PySpark 作业、Airflow DAG、数据质量检查 |
| | SQL | 数据查询、转换、建模、优化（核心技能） |
| | Java/Scala | Spark/Flink 大规模数据处理作业 |
| | Shell | 数据管道自动化脚本、运维操作 |
| **大数据框架** | Apache Spark | 编写 PySpark/Scala 批处理与流处理作业，调优 shuffle/内存/并行度 |
| | Apache Flink | 编写 Flink SQL/DataStream 实时作业，配置 Checkpoint/Savepoint |
| | Apache Kafka | Topic 设计、分区策略、生产/消费配置、Kafka Connect |
| | Hadoop (HDFS) | 文件存储管理、NameNode/DataNode 排障 |
| | Apache Hive | HQL 查询、分区表、ORC/Parquet 格式优化 |
| **数据仓库/湖** | Snowflake / BigQuery / Redshift | SQL 查询优化、物化视图、自动扩缩容配置 |
| | ClickHouse / Apache Doris | OLAP 查询优化、MergeTree 引擎选择、分区策略 |
| | Delta Lake / Iceberg / Hudi | 表格式管理、时间旅行、ACID 事务、Compaction |
| | Apache Hudi | 增量查询、Upsert/Delete、Clustering |
| **编排调度** | Apache Airflow | 编写 DAG、配置 Sensor/Pool/SLA、任务重试与告警 |
| | Dagster | Asset-based 数据管道、资源管理、launchpad |
| | Prefect | 工作流定义、自动重试、通知 |
| | Apache DolphinScheduler | 分布式任务调度、工作流定义 |
| **流处理** | Apache Kafka | Topic 设计、分区策略、生产/消费调优、Kafka Streams |
| | Apache Flink | Flink SQL / DataStream 作业、窗口聚合、状态管理 |
| | Spark Streaming | Structured Streaming 实时管道 |
| **数据转换** | dbt | 编写 SQL 模型、测试、文档生成、增量策略 |
| | Great Expectations | 数据质量期望定义、验证、报告 |
| | dbt test | 数据完整性、唯一性、引用完整性测试 |
| **数据湖格式** | Delta Lake | 表管理、时间旅行、Z-Order 优化、Vacuum |
| | Apache Iceberg | 表格式管理、分区演进、隐藏分区 |
| | Apache Hudi | Upsert/Delete、增量查询、Clustering |
| **编排调度** | Apache Airflow | 编写 DAG、配置 Sensor/Pool/SLA、任务重试与告警 |
| | Dagster | Asset-based 管道、资源管理、launchpad |
| | Apache DolphinScheduler | 可视化工作流定义、任务依赖 |
| **数据查询** | Trino / Presto | 联邦查询、跨数据源分析 |
| | Apache Hive | HQL 查询、分区表、存储格式优化 |
| | ClickHouse | OLAP 查询优化、物化视图、TTL |
| **消息队列** | Apache Kafka | Topic 设计、分区策略、生产/消费配置、Kafka Streams |
| | RabbitMQ / Pulsar | 消息队列配置、路由策略 |
| **数据湖格式** | Delta Lake | 表管理、时间旅行、Z-Order、Vacuum、Optimize |
| | Apache Iceberg | 表格式管理、分区演进、隐藏分区、Compaction |
| | Apache Hudi | Upsert/Delete、增量查询、Clustering、Cleaner |
| **编排调度** | Apache Airflow | DAG 编写、Sensor/Pool/SLA、任务重试与告警、动态 DAG |
| | Dagster | Asset 定义、资源管理、launchpad、代码位置 |
| | Prefect | 工作流定义、自动重试、并发控制 |
| **数据转换** | dbt | 模型编写、test、文档生成、增量策略、snapshot |
| | dbt test | 唯一性、非空、引用完整性、自定义测试 |
| **数据质量** | Great Expectations | Expectation Suite 定义、数据验证、Data Docs 生成 |
| | Soda Core | 数据质量扫描、异常检测 |
| | Deequ | Spark 上的数据质量验证（AWS 开源） |
| **数据湖格式** | Delta Lake | 表管理、时间旅行、Z-Order 优化、Vacuum、Optimize |
| | Apache Iceberg | 表格式管理、分区演进、隐藏分区、Compaction、Snapshot 管理 |
| | Apache Hudi | Upsert/Delete、增量查询、Clustering、Cleaner、Compaction |
| **编排调度** | Apache Airflow | DAG 编写、Sensor/Pool/SLA、任务重试与告警、动态 DAG 生成 |
| | Dagster | Asset 定义、资源管理、代码位置、传感器、调度 |
| | Prefect | 工作流定义、自动重试、并发控制、通知 |
| | Apache DolphinScheduler | 可视化工作流定义、任务依赖、告警 |
| **流处理** | Apache Kafka | Topic 设计、分区策略、生产/消费配置、Kafka Connect、Schema Registry |
| | Apache Flink | Flink SQL / DataStream API、窗口聚合、状态管理、Checkpoint/Savepoint |
| | Spark Structured Streaming | 微批处理、连续处理、Watermark、输出模式 |
| **数据转换** | dbt | 模型编写（SQL/Python）、test、文档、增量策略、snapshot、exposure |
| | dbt Cloud / dbt Core | 项目配置、CI/CD、文档托管 |
| **数据质量** | Great Expectations | Expectation Suite 定义、数据验证、Data Docs、Checkpoint |
| | Soda Core | 数据质量扫描、异常检测、指标监控 |
| | Deequ | Spark 上的数据质量约束验证 |
| **数据湖格式** | Delta Lake | CREATE TABLE USING delta、Z-ORDER BY、OPTIMIZE、VACUUM、DESCRIBE HISTORY |
| | Apache Iceberg | CREATE TABLE USING iceberg、分区演进、Snapshot 管理、Compaction |
| | Apache Hudi | COW/MOR 表类型、Upsert/Delete、Incremental Query、Clustering |
| **数据查询** | Trino / Presto | 联邦查询、跨数据源 JOIN、连接池配置 |
| | Apache Hive | HQL 查询、分区表、ORC/Parquet 格式、CBO 优化 |
| | ClickHouse | MergeTree 引擎选择、ORDER BY/BY 设计、物化视图、TTL |
| | Apache Doris | 明细/聚合/Unique 模型选择、Rollup 表、Colocation Join |
| **消息队列** | Apache Kafka | Topic 设计、分区策略、副本配置、生产/消费参数调优 |
| | Kafka Connect | Source/Sink Connector 配置、单模式/分布式模式 |
| | Schema Registry | Avro/Protobuf/JSON Schema 管理、兼容性策略 |
| **数据转换** | dbt | 模型编写（SQL/Python）、test、文档、增量策略、snapshot、exposure |
| | dbt Cloud | 项目配置、CI/CD、文档托管、IDE |
| **数据质量** | Great Expectations | Expectation Suite 定义、Checkpoint、Data Docs、Profiler |
| | Soda Core | 数据质量扫描、异常检测、指标监控、Soda Cloud |
| | Deequ | Spark 上的数据质量约束验证、指标计算、异常检测 |
| **数据湖格式** | Delta Lake | CREATE TABLE USING delta、Z-ORDER BY、OPTIMIZE、VACUUM、DESCRIBE HISTORY |
| | Apache Iceberg | CREATE TABLE USING iceberg、分区演进、Snapshot 管理、Expire Snapshots |
| | Apache Hudi | COW/MOR 表类型、Upsert/Delete、Incremental Query、Clustering、Cleaner |
| **编排调度** | Apache Airflow | DAG 编写（PythonOperator/BashOperator/Sensor）、Pool/SLA、重试策略、告警 |
| | Dagster | Asset 定义、资源管理、传感器、调度、代码位置、launchpad |
| | Prefect | Flow/Task 定义、自动重试、并发控制、通知、部署 |
| **数据查询** | Trino / Presto | 联邦查询、跨数据源 JOIN、连接池配置、查询优化 |
| | Apache Hive | HQL 查询、分区表、ORC/Parquet 格式、CBO 优化 |
| | ClickHouse | MergeTree 引擎选择、ORDER BY 设计、物化视图、TTL、分布式表 |
| | Apache Doris | 明细/聚合/Unique 模型、Rollup 表、Colocation Join、Bucket 分桶 |
| **消息队列** | Apache Kafka | Topic 设计、分区/副本配置、生产/消费参数调优、JMX 监控 |
| | Kafka Connect | Source/Sink Connector 配置、单机/分布式模式、REST API |
| | Schema Registry | Avro/Protobuf/JSON Schema 注册、兼容性检查 |
| **数据转换** | dbt | 模型编写（SQL/Python）、test、文档、增量策略、snapshot、exposure、hooks |
| | dbt Cloud | 项目配置、CI/CD、文档托管、IDE、job 调度 |
| **数据质量** | Great Expectations | Expectation Suite 定义、Checkpoint、Data Docs、Profiler、Action |
| | Soda Core | 数据质量扫描、异常检测、指标监控、Soda Cloud 集成 |
| | Deequ | Spark 上的数据质量约束验证、指标计算、异常检测 |
| **数据湖格式** | Delta Lake | CREATE TABLE USING delta、Z-ORDER BY、OPTIMIZE、VACUUM、DESCRIBE HISTORY |
| | Apache Iceberg | CREATE TABLE USING iceberg、分区演进、Snapshot 管理、Expire Snapshots、Rewrite |
| | Apache Hudi | COW/MOR 表类型、Upsert/Delete、Incremental Query、Clustering、Cleaner、Compaction |
| **数据查询** | Trino / Presto | 联邦查询、跨数据源 JOIN、连接池配置、查询优化 |
| | Apache Hive | HQL 查询、分区表、ORC/Parquet 格式、CBO 优化 |
| | ClickHouse | MergeTree 引擎选择、ORDER BY 设计、物化视图、TTL、分布式表、跳数索引 |
| | Apache Doris | 明细/聚合/Unique 模型、Rollup 表、Colocation Join、Bucket 分桶、物化视图 |
| **数据编排** | Apache Airflow | DAG 编写（PythonOperator/BashOperator/Sensor/ExternalTaskSensor）、Pool/SLA、重试策略、告警、动态 DAG |
| | Dagster | Asset 定义、资源管理、传感器、调度、代码位置、launchpad、I/O 管理器 |
| | Prefect | Flow/Task 定义、自动重试、并发控制、通知、部署、work pool |
| **数据转换** | dbt | 模型编写（SQL/Python）、test、文档、增量策略、snapshot、exposure、hooks、macro |
| | dbt Cloud | 项目配置、CI/CD、文档托管、IDE、job 调度、环境管理 |
| **数据质量** | Great Expectations | Expectation Suite 定义、Checkpoint、Data Docs、Profiler、Action（通知/告警） |
| | Soda Core | 数据质量扫描、异常检测、指标监控、Soda Cloud 集成 |
| | Deequ | Spark 上的数据质量约束验证、指标计算、异常检测、建议约束 |
| **数据湖格式** | Delta Lake | CREATE TABLE USING delta、Z-ORDER BY、OPTIMIZE、VACUUM、DESCRIBE HISTORY、CONVERT TO DELTA |
| | Apache Iceberg | CREATE TABLE USING iceberg、分区演进、Snapshot 管理、Expire Snapshots、Rewrite Data Files |
| | Apache Hudi | COW/MOR 表类型、Upsert/Delete、Incremental Query、Clustering、Cleaner、Compaction |
| **数据查询** | Trino / Presto | 联邦查询、跨数据源 JOIN、连接池配置、查询优化、资源组 |
| | Apache Hive | HQL 查询、分区表、ORC/Parquet 格式、CBO 优化、Vectorization |
| | ClickHouse | MergeTree 引擎选择、ORDER BY 设计、物化视图、TTL、跳数索引、分布式表 |
| | Apache Doris | 明细/聚合/Unique 模型、Rollup 表、Colocation Join、Bucket 分桶、物化视图 |
| **数据同步** | Apache SeaTunnel | 多数据源同步、插件化 Connector、Flink/Zeta 引擎 |
| | DataX / Addax | 异构数据源离线同步、全量/增量模式 |
| | Canal / Debezium | MySQL/PostgreSQL CDC、Binlog 解析、实时同步 |
| | Maxwell | MySQL Binlog 解析、Kafka 输出、DDL 同步 |
| **监控与可观测性** | Grafana + Prometheus | 数据管道监控 Dashboard、告警规则、指标采集 |
| | OpenLineage | 数据血缘采集、Airflow/Flink/Spark 集成 |
| | DataHub / Apache Atlas | 元数据管理、数据发现、血缘可视化 |
| | Marquez | 数据血缘追踪、作业依赖可视化 |
| **基础设施** | Docker / Docker Compose | 数据服务容器化部署（Kafka/Spark/Airflow/ClickHouse） |
| | Kubernetes | 大数据集群容器化部署、Operator 管理（Strimzi/Flink K8s Operator） |
| | Terraform | 数据基础设施即代码（云上数据服务） |
| | Ansible | 大数据集群自动化部署与配置 |
## 典型场景与工作流
### 场景1：搭建 MySQL → 数据仓库 ETL 管道
用户："帮我搭建一个从 MySQL 到 ClickHouse 的 ETL 管道，每天增量同步"
│
▼
- 需求确认
├─ 数据源：MySQL 表结构、数据量、增量字段（update_time / id）
├─ 目标：ClickHouse 表结构、分区键、排序键
├─ 时效性：T+1 天级 / 小时级
└─ 数据量：日增多少行、总数据量
│
▼
- 方案设计
├─ 架构：MySQL CDC (Canal) → Kafka → Flink → ClickHouse
│ 或：Airflow 调度 → Spark/Python ETL → ClickHouse
├─ 增量策略：时间戳增量 / Binlog CDC
├─ 数据模型：ODS → DWD → DWS → ADS 分层
└─ 容错：断点续传、幂等写入
│
▼
- 执行
├─ 编写 ETL 脚本（PySpark / Python + pandas）
├─ 配置 Airflow DAG（调度、重试、告警）
├─ 创建目标表（分区键、排序键、TTL）
├─ 运行管道
└─ 验证数据（行数对账、字段抽样）
│
▼
- 交付
├─ DAG 代码 + ETL 脚本
├─ 数据字典（字段说明、血缘关系）
├─ 监控 Dashboard（管道延迟、数据量趋势）
└─ 运维手册（重跑、排障、扩缩容）
### 场景2：实时流处理管道
用户："帮我搭建一个实时用户行为分析管道，Kafka → Flink → ClickHouse"
│
▼
- 需求确认
├─ 数据源：用户行为事件（埋点日志）、消息格式（JSON/Avro/Protobuf）
├─ 时效性：秒级延迟
├─ 计算逻辑：PV/UV、漏斗分析、留存分析
└─ 数据量：每秒多少条事件
│
▼
- 方案设计
├─ 架构：Kafka（事件总线）→ Flink（实时计算）→ ClickHouse（OLAP 存储）
├─ Topic 设计：按事件类型分区、合理分区数
├─ Flink 作业：窗口聚合（Tumble/Hop/Session）、状态管理
└─ 容错：Kafka 副本 + Flink Checkpoint + Exactly-Once
│
▼
- 执行
├─ 创建 Kafka Topic（分区数、副本数、保留策略）
├─ 编写 Flink SQL 作业（CREATE TABLE → INSERT INTO）
├─ 配置 Checkpoint（间隔、模式、存储后端）
├─ 创建 ClickHouse 目标表（ReplicatedMergeTree、物化视图）
├─ 部署作业
└─ 验证：Kafka 生产测试数据 → 检查 ClickHouse 结果
│
▼
- 交付
├─ Flink 作业代码 + SQL
├─ Kafka Topic 配置说明
├─ ClickHouse 表结构 DDL
├─ 监控 Dashboard（Kafka Lag、Flink Checkpoint、ClickHouse QPS）
└─ 排障手册（数据延迟、反压、OOM 处理）
### 场景3：Airflow 数据管道编排
用户："帮我用 Airflow 编排一个每日数据管道，从多个 MySQL 同步到 Hive"
│
▼
- 需求确认
├─ 数据源：多个 MySQL 实例、表结构、增量字段
├─ 目标：Hive 分区表（按日期分区）
├─ 调度频率：每天凌晨 2 点
└─ 依赖关系：表 A 完成后才能跑表 B
│
▼
- DAG 设计
├─ start → check_partition_sensor → parallel_extract → transform → load → quality_check → finish
├─ 并行抽取多个 MySQL 表
├─ 数据转换（清洗、去重、类型转换）
├─ 数据质量检查（行数对账、空值检查）
└─ 失败告警（钉钉/企微/邮件）
│
▼
- 执行
├─ 编写 DAG（PythonOperator + BashOperator + Sensor）
├─ 配置连接（MySQL、Hive 连接信息）
├─ 设置 Pool（控制并行度）
├─ 配置告警（on_failure_callback）
├─ 部署 DAG
└─ 触发测试运行
│
▼
- 交付
├─ DAG 代码
├─ 数据字典
├─ 调度配置说明
└─ 排障手册
### 场景4：Spark 作业开发与调优
用户："帮我写一个 Spark 作业处理用户行为日志，并做性能调优"
│
▼
- 需求确认
├─ 输入：HDFS/S3 上的 Parquet/JSON 日志文件
├─ 计算逻辑：用户会话聚合、行为路径分析
├─ 输出：聚合后的 Parquet 表
└─ 数据量：日增 10 亿条
│
▼
- 方案设计
├─ 读取：spark.read.parquet / spark.read.json
├─ 转换：groupBy + window + agg、UDF
├─ 写入：partitionBy + format parquet + mode overwrite
└─ 调优：合理并行度、broadcast join、数据倾斜处理
│
▼
- 执行
├─ 编写 PySpark 作业
├─ 配置 SparkSession（executor 内存/核数、shuffle 分区数、动态分配）
├─ 处理数据倾斜（salting、broadcast hash join、AQE）
├─ 提交作业（spark-submit）
└─ 验证结果
│
▼
- 调优检查清单
├─ [ ] 合理并行度（executor 核数 × 数量 × 2~3）
├─ [ ] 数据倾斜处理（salting / broadcast join / AQE）
├─ [ ] shuffle 优化（减少 shuffle 数据量、使用 Tungsten）
├─ [ ] 文件大小控制（maxRecordsPerFile / coalesce / repartition）
├─ [ ] 序列化（Kryo 替代 Java 序列化）
├─ [ ] 缓存策略（cache/persist 级别选择）
├─ [ ] 动态资源分配（dynamicAllocation）
└─ [ ] AQE（Adaptive Query Execution）开启
### 场景4：dbt 数据转换项目
用户："帮我用 dbt 搭建数据转换层，从原始数据到分析层"
│
▼
- 项目初始化
├─ dbt init project_name
├─ 配置 profiles.yml（连接目标数仓）
└─ 配置 dbt_project.yml（模型路径、目录结构）
│
▼
- 模型分层
├─ staging/：原始数据清洗、类型转换、列重命名
├─ intermediate/：中间聚合、业务逻辑计算
├─ marts/：主题域宽表、聚合指标
└─ 模型依赖：{{ ref('stg_orders') }} 自动构建血缘
│
▼
- 测试与文档
├─ schema.yml：定义列约束（unique、not_null、accepted_values）
├─ dbt test：运行测试、查看结果
├─ dbt docs generate：生成文档站点
└─ dbt docs serve：本地预览
│
▼
- 增量策略
├─ 配置 materialized='incremental'
├─ unique_key + merge 策略
├─ on_schema_change 处理
└─ 增量回刷
### 场景5：数据质量保障
用户："帮我搭建数据质量监控体系"
│
▼
- 方案设计
├─ 工具选择：Great Expectations / Soda / Deequ
├─ 检查维度：完整性、准确性、一致性、及时性、唯一性
├─ 检查频率：每次管道运行 + 定时全量扫描
└─ 告警方式：钉钉/企微/邮件/Slack
│
▼
- 执行（以 Great Expectations 为例）
├─ great_expectations init
├─ 配置 Data Source（数据库/文件/数据湖）
├─ 创建 Expectation Suite
│ ├─ expect_column_values_to_not_be_null
│ ├─ expect_column_values_to_be_unique
│ ├─ expect_column_values_to_be_between
│ ├─ expect_table_row_count_to_be_between
│ └─ expect_column_pair_values_to_be_equal
├─ 配置 Checkpoint（数据源 + Suite + Action）
├─ 配置 Action（通知、写入结果表）
└─ 生成 Data Docs
│
▼
- 集成到管道
├─ Airflow DAG 中插入 GreatExpectationsOperator
├─ 质量检查失败时触发告警 + 暂停下游任务
└─ 定期生成数据质量报告
### 场景5：数据倾斜排查与优化
用户："Spark 作业跑得很慢，怀疑数据倾斜，帮我排查"
│
▼
- 确认现象
├─ Spark UI：某些 Task 运行时间远长于其他 Task
├─ 某些 Executor 处理数据量远大于其他 Executor
└─ Shuffle 阶段某些分区数据量巨大
│
▼
- 定位倾斜
├─ Spark UI → Stages → 查看 Task 数据分布
├─ 查看 Shuffle Read/Write 大小
├─ 对 key 做 count groupBy 检查分布
└─ 常见倾斜场景：join key 空值、热点 key、分区键不均匀
│
▼
- 优化方案
├─ Salting：给热点 key 加随机前缀，分散到多个分区
├─ Broadcast Hash Join：小表广播，避免 shuffle
├─ AQE（Adaptive Query Execution）：自动合并小分区、skew join 优化
├─ 增加 shuffle 分区数：spark.sql.shuffle.partitions
└─ 空值处理：过滤或随机分散空值 key
│
▼
- 验证
├─ 重新运行作业
├─ 对比 Task 时间分布
└─ 确认整体运行时间下降
### 场景6：实时流处理管道
用户："帮我搭建一个实时用户行为分析管道，Kafka → Flink → ClickHouse"
│
▼
- 需求确认
├─ 数据源：用户行为事件（埋点日志）、消息格式（JSON/Avro/Protobuf）
├─ 时效性：秒级延迟
├─ 计算逻辑：PV/UV、漏斗分析、留存分析
└─ 数据量：每秒多少条事件
│
▼
- 方案设计
├─ 架构：Kafka（事件总线）→ Flink（实时计算）→ ClickHouse（OLAP 存储）
├─ Topic 设计：按事件类型分区、合理分区数（≥ 消费并行度）
├─ Flink 作业：窗口聚合（Tumble/Hop/Session）、状态管理
└─ 容错：Kafka 副本 + Flink Checkpoint + Exactly-Once
│
▼
- 执行
├─ 创建 Kafka Topic（分区数、副本数、保留策略、清理策略）
├─ 编写 Flink SQL 作业
│ ├─ CREATE TABLE source (Kafka 连接器)
│ ├─ CREATE VIEW agg_view (窗口聚合)
│ └─ INSERT INTO sink (ClickHouse 连接器)
├─ 配置 Checkpoint（间隔、模式、存储后端）
├─ 创建 ClickHouse 目标表（ReplicatedMergeTree、物化视图）
├─ 部署作业
└─ 验证：Kafka 生产测试数据 → 检查 ClickHouse 结果
│
▼
- 交付
├─ Flink 作业代码 + SQL
├─ Kafka Topic 配置说明
├─ ClickHouse 表结构 DDL
├─ 监控 Dashboard（Kafka Lag、Flink Checkpoint、ClickHouse QPS）
└─ 排障手册（反压、OOM、数据延迟处理）
### 场景6：数据湖搭建
用户："帮我搭建数据湖，管理 PB 级数据"
│
▼
- 需求确认
├─ 数据源类型：结构化/半结构化/非结构化
├─ 数据量级：当前/未来增长
├─ 查询模式：OLAP / 机器学习 / 即席查询
└─ 技术栈偏好：开源/云原生
│
▼
- 方案设计
├─ 存储：HDFS / S3 / MinIO / OSS
├─ 表格式：Delta Lake / Iceberg / Hudi（三选一或混合）
├─ 计算引擎：Spark / Trino / Flink
├─ 目录服务：Hive Metastore / AWS Glue / Nessie
└─ 数据分层：Bronze → Silver → Gold（Medallion Architecture）
│
▼
- 执行
├─ 配置存储后端（HDFS 集群 / S3 Bucket）
├─ 配置 Hive Metastore（PostgreSQL 后端）
├─ 创建数据湖表（Delta/Iceberg/Hudi 格式）
├─ 配置数据入湖管道（Spark Structured Streaming / Kafka Connect）
├─ 配置 Compaction 策略（自动合并小文件）
└─ 配置数据保留策略（TTL / Snapshot Expire）
│
▼
- 交付
├─ 数据湖架构图
├─ 表格式配置说明
├─ 入湖管道代码
├─ 数据保留与清理策略
└─ 运维手册
## 性能优化检查清单
### Spark 作业调优
- [ ] 合理并行度：executor 核数 × executor 数量 × 2~3
- [ ] 数据倾斜处理：salting / broadcast join / AQE skew join
- [ ] Shuffle 优化：减少 shuffle 数据量、使用 Tungsten 排序
- [ ] 序列化：Kryo 序列化（spark.serializer）
- [ ] 文件大小：每个输出文件 128MB~1GB
- [ ] 动态资源分配：spark.dynamicAllocation.enabled
- [ ] AQE 开启：spark.sql.adaptive.enabled = true
- [ ] 广播阈值：spark.sql.autoBroadcastJoinThreshold
- [ ] 内存配置：executor 内存比例（spark.memory.fraction / spark.memory.storageFraction）
### SQL 查询优化
- [ ] EXPLAIN ANALYZE 分析执行计划
- [ ] 索引使用（B-tree / Bitmap / Hash / GIN）
- [ ] 分区裁剪（Partition Pruning）
- [ ] 避免 SELECT *，只取需要的列
- [ ] 避免 N+1 查询，使用 JOIN 或子查询
- [ ] 大表 JOIN 使用 Broadcast Join 或 Bucket Map Join
- [ ] 物化视图加速聚合查询
- [ ] 避免函数包裹索引列（WHERE DATE(create_time) = '2024-01-01' → WHERE create_time >= '2024-01-01' AND create_time < '2024-01-02'）
### 数据管道优化
- [ ] 合理设置批处理大小（batch size / spark.sql.broadcastTimeout）
- [ ] 小文件合并（coalesce / repartition / OPTIMIZE）
- [ ] 压缩算法选择（Snappy / Zstd / LZ4 / Gzip）
- [ ] 列裁剪（只读取需要的列）
- [ ] 分区裁剪（只读取需要的分区）
- [ ] 谓词下推（Predicate Pushdown）
- [ ] 向量化读取（spark.sql.parquet.enableVectorizedReader）
## 数据建模规范
### 分层设计
| 层级 | 名称 | 说明 | 数据粒度 |
|------|------|------|---------|
| ODS | 操作数据层 | 原始数据，与源系统一致 | 与源系统一致 |
| DWD | 明细数据层 | 清洗、去重、标准化后的明细数据 | 业务明细 |
| DWS | 汇总数据层 | 按主题域轻度汇总 | 汇总粒度 |
| ADS | 应用数据层 | 面向业务应用的个性化数据 | 业务需求粒度 |
| DIM | 维度层 | 公共维度表（日期、用户、产品） | 维度属性 |
### 命名规范
- 表名：`{层级}_{主题域}_{表名}`（如 `dwd_trade_order_detail`）
- 字段：snake_case，主键 `id`，外键 `{表名}_id`
- 分区字段：`dt`（日期分区，格式 yyyy-MM-dd）
- 时间字段：`create_time`、`update_time`
### 存储格式选择
| 格式 | 适用场景 | 压缩比 | 查询性能 |
|------|---------|--------|---------|
| **Parquet** | OLAP 查询、列存分析 | 高 | 高 |
| **ORC** | Hive 查询、ACID 事务 | 高 | 高 |
| **Avro** | 流处理、Kafka 消息 | 中 | 中 |
| **Delta** | 数据湖、ACID、时间旅行 | 高 | 高 |
| **JSON** | 日志、半结构化数据 | 低 | 低 |
## 数据质量检查清单
- [ ] **完整性**：必填字段无 NULL、行数在合理范围内
- [ ] **准确性**：数值在合理范围、格式正确（日期/邮箱/手机号）
- [ ] **一致性**：关联表之间的外键约束、枚举值一致
- [ ] **唯一性**：主键无重复、业务唯一键无重复
- [ ] **及时性**：数据到达时间在 SLA 范围内
- [ ] **完整性**：无缺失分区、无断流
## 常用命令速查
```bash
# Spark
spark-submit --master yarn --deploy-mode cluster --num-executors 100 --executor-cores 4 --executor-memory 8g job.py
spark-sql -e "SELECT * FROM table"
spark-shell --conf spark.sql.adaptive.enabled=true
# Flink
flink run -m yarn-cluster -d -p 10 job.jar
flink list -m yarn-cluster
flink cancel -m yarn-cluster <job_id>
flink savepoint <job_id> hdfs:///flink/savepoints
# Kafka
kafka-topics.sh --create --topic my_topic --partitions 6 --replication-factor 3 --bootstrap-server localhost:9092
kafka-console-producer.sh --topic my_topic --bootstrap-server localhost:9092
kafka-console-consumer.sh --topic my_topic --from-beginning --bootstrap-server localhost:9092
kafka-consumer-groups.sh --group my_group --describe --bootstrap-server localhost:9092
# Airflow
airflow dags list
airflow dags trigger my_dag
airflow tasks test my_dag task_name 2024-01-01
airflow dags backfill my_dag -s 2024-01-01 -e 2024-01-07
# dbt
dbt run --models +my_model
dbt test --select my_model
dbt docs generate
dbt docs serve
dbt run --full-refresh --select my_model
# Great Expectations
great_expectations checkpoint run my_checkpoint
great_expectations docs build
great_expectations suite new
# ClickHouse
SELECT * FROM system.parts WHERE table = 'my_table'
OPTIMIZE TABLE my_table FINAL
ALTER TABLE my_table DELETE WHERE dt < '2024-01-01'
# HDFS
hdfs dfs -ls /user/hive/warehouse/
hdfs dfs -du -h /user/hive/warehouse/
hdfs dfsadmin -report
常见问题排查
| 问题类型 | 排查步骤 |
|---|
| Spark OOM | Spark UI → Executors 页 → 查看 GC 时间/内存使用 → 检查数据倾斜 → 调整 spark.memory.fraction / off-heap |
| Kafka 消费延迟 | kafka-consumer-groups --describe → 查看 LAG → 检查消费者处理能力 → 增加分区/消费者 |
| Flink 反压 | Flink UI → 查看反压状态 → 定位瓶颈算子 → 优化并行度/算子链 |
| Airflow 任务卡住 | 查看 task 日志 → 检查资源池 → 检查数据库连接 → 手动 kill 重跑 |
| HDFS 空间不足 | hdfs dfs -du -h → 定位大目录 → 清理过期数据/调整副本数 |
| ClickHouse 查询慢 | EXPLAIN → 查看是否走索引 → 检查 ORDER BY 设计 → 检查 MergeTree 碎片 |
| 数据倾斜 | Spark UI → 查看 Task 数据分布 → groupBy key 检查 → salting / broadcast join |
| 小文件过多 | 检查 Spark 输出文件数 → coalesce/repartition → 设置 maxRecordsPerFile → OPTIMIZE |
数据治理清单
数据建模规范
星型模型
事实表（交易事实）
├── 维度：时间维度（日期）
├── 维度：产品维度（产品ID → 产品名称/品类/品牌）
├── 维度：用户维度（用户ID → 用户属性）
└── 维度：门店维度（门店ID → 门店属性）
分层 SQL 示例
CREATE TABLE ods_trade_order (
order_id BIGINT,
user_id BIGINT,
product_id BIGINT,
amount DECIMAL(10,2),
status STRING,
create_time TIMESTAMP,
update_time TIMESTAMP
) PARTITIONED BY (dt STRING)
STORED AS PARQUET;
CREATE TABLE dwd_trade_order_detail
USING delta
PARTITIONED BY (dt)
AS SELECT
order_id,
user_id,
product_id,
amount,
status,
create_time,
update_time,
CASE WHEN status = 'paid' THEN 1 ELSE 0 END AS is_paid,
CASE WHEN status = 'cancelled' THEN 1 ELSE 0 END AS is_cancelled
FROM ods_trade_order
WHERE dt = '${dt}';
CREATE TABLE dws_trade_daily (
dt STRING,
total_orders BIGINT,
total_amount DECIMAL(15,2),
paid_orders BIGINT,
paid_amount DECIMAL(15,),
unique_users
) STORED PARQUET;
常用代码片段
PySpark ETL 模板
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, when, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, LongType, DecimalType, TimestampType
spark = SparkSession.builder \
.appName("etl_mysql_to_clickhouse") \
.config("spark.sql.adaptive.enabled", "true") \
.config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
.config("spark.sql.adaptive.skewJoin.enabled", "true") \
.getOrCreate()
df = spark.read \
.format("jdbc") \
.option("url", "jdbc:mysql://host:3306/db") \
.option("dbtable", "table_name") \
.option("user", "user") \
.option("password", "pass") \
.option("numPartitions", 10) \
.option("partitionColumn", "id") \
.option("lowerBound", 1) \
.option("upperBound", 10000000) \
.load()
df_clean = df \
.dropDuplicates(["order_id"]) \
.filter(col("amount").isNotNull()) \
.withColumn("dt", col("create_time").cast("date")) \
.withColumn("is_valid", when(col("amount") > 0, True).otherwise(False))
df_clean.write \
.() \
.mode() \
.partitionBy() \
.option(, ) \
.save()
Airflow DAG 模板
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.providers.apache.hive.hooks.hive import HiveHook
from airflow.utils.dates import days_ago
default_args = {
'owner': 'data_team',
'depends_on_past': False,
'email_on_failure': True,
'email': ['oncall@company.com'],
'retries': 3,
'retry_delay': timedelta(minutes=5),
'retry_exponential_backoff': True,
'execution_timeout': timedelta(hours=2),
}
with DAG(
dag_id='etl_mysql_to_clickhouse',
default_args=default_args,
description='每日从 MySQL 同步数据到 ClickHouse',
schedule_interval='0 2 * * *',
start_date=days_ago(1),
catchup=False,
tags=['etl', 'clickhouse'],
) as dag:
check_partition = ExternalTaskSensor(
task_id='check_upstream_partition',
external_dag_id='upstream_dag',
external_task_id='done',
timeout=,
poke_interval=,
mode=,
)
extract_mysql = PythonOperator(
task_id=,
python_callable=extract_mysql_data,
op_kwargs={: , : },
pool=,
)
transform = PythonOperator(
task_id=,
python_callable=transform_data,
)
load_clickhouse = PythonOperator(
task_id=,
python_callable=load_to_clickhouse,
)
quality_check = PythonOperator(
task_id=,
python_callable=run_quality_checks,
)
check_partition >> extract_mysql >> transform >> load_clickhouse >> quality_check
Flink SQL 实时聚合模板
CREATE TABLE source_events (
event_id STRING,
user_id STRING,
event_type STRING,
page_url STRING,
event_time TIMESTAMP(3),
metadata ROW<device STRING, ip STRING>,
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'user_events',
'properties.bootstrap.servers' = 'kafka:9092',
'properties.group.id' = 'flink_consumer',
'format' = 'json',
'scan.startup.mode' = 'latest-offset'
);
CREATE VIEW pv_uv_5min AS
SELECT
TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start,
TUMBLE_END(event_time, INTERVAL '5' MINUTE) AS window_end,
page_url,
COUNT(*) AS pv,
COUNT(DISTINCT user_id) AS uv
source_events
TUMBLE(event_time, ),
page_url;
sink_pv_uv (
window_start (),
window_end (),
page_url STRING,
pv ,
uv
) (
,
,
,
,
,
);
sink_pv_uv
window_start, window_end, page_url, pv, uv
pv_uv_5min;
dbt 模型示例
WITH source AS (
SELECT * FROM {{ source('mysql', 'orders') }}
),
renamed AS (
SELECT
id AS order_id,
user_id,
product_id,
amount,
status,
created_at AS create_time,
updated_at AS update_time
FROM source
WHERE amount IS NOT NULL
)
SELECT * FROM renamed
SELECT
product_id,
product_name,
category,
brand,
price,
created_at
FROM {{ ref('stg_products') }}
WHERE is_active = true
{{ config(
materialized='incremental',
unique_key='order_id',
incremental_strategy='merge',
on_schema_change='sync_all_columns'
) }}
SELECT
o.order_id,
o.user_id,
o.product_id,
o.amount,
o.status,
o.create_time,
o.update_time,
p.category,
p.brand
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref() }} p o.product_id p.product_id
{ if is_incremental() }
o.update_time ( (update_time) {{ this }})
{ endif }
与现有技能的复用关系
| 现有技能 | 复用方式 |
|---|
arch-c4-diagram | 需要画数据架构图时加载 |
arch-adr | 需要记录数据架构决策时加载 |
arch-tech-evaluation | 需要技术选型评估时加载 |
ops-engineer | 需要部署/运维数据基础设施时加载 |
backend-developer | 需要开发数据服务 API 时加载 |
chinese-pdf-generation | 需要输出数据字典/架构文档为 PDF 时加载 |
典型对话示例
示例1：搭建 ETL 管道
用户："帮我搭建一个从 MySQL 到 ClickHouse 的 ETL 管道"
Agent：确认数据源/目标/量级 → 方案设计 → 编写 ETL 脚本 → 配置 Airflow DAG → 运行 → 验证 → 交付
示例2：实时流处理
用户："帮我搭建 Kafka → Flink → ClickHouse 实时分析管道"
Agent：确认事件格式/计算逻辑 → 设计 Topic → 编写 Flink SQL → 创建 ClickHouse 表 → 部署 → 验证
示例3：Spark 作业调优
用户："Spark 作业跑得慢，帮我优化"
Agent：查看 Spark UI → 定位瓶颈 → 检查数据倾斜 → 调整参数 → 验证性能提升
示例4：dbt 项目搭建
用户："帮我用 dbt 搭建数据转换层"
Agent：初始化项目 → 分层设计 → 编写模型 → 配置测试 → 生成文档 → 配置增量策略
示例5：数据质量监控
用户："帮我搭建数据质量监控"
Agent：选择工具 → 定义质量规则 → 配置 Checkpoint → 集成到管道 → 配置告警 → 生成报告
示例6：数据湖搭建
用户："帮我搭建数据湖"
Agent：确认存储/计算/表格式 → 设计分层 → 配置表格式 → 入湖管道 → 配置 Compaction/清理 → 交付
输出规范
| 输出类型 | 格式 | 说明 |
|---|
| 数据管道代码 | Python/SQL/Java | ETL/ELT 脚本、Spark/Flink 作业 |
| 调度配置 | Python/YAML | Airflow DAG、Dagster Job、Prefect Flow |
| 数据模型 | SQL | 建表 DDL、dbt 模型、视图定义 |
| 数据字典 | Markdown/HTML | 表结构、字段说明、血缘关系 |
| 架构图 | Mermaid | 数据管道架构、数据流图 |
| 配置文件 | YAML/TOML | Spark 配置、Kafka 配置、dbt 配置 |
| 质量报告 | Markdown/HTML | 数据质量检查结果、异常报告 |
| 运维手册 | Markdown | 管道运维、排障、扩缩容指南 |
