aiskillstore 数据工程师 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 |
数据治理清单
- 元数据管理：表注释、字段注释、数据字典
- 数据血缘：记录数据从哪里来、经过哪些转换、到哪里去
- 数据分类：敏感数据识别与分级（PII、财务、业务）
- 数据脱敏：手机号/身份证/邮箱脱敏规则
- 权限管理：表级/行级/列级权限控制
- 数据生命周期：冷热数据分层、归档策略、清理策略
- 数据备份：关键表定期备份、可恢复性验证
- 变更管理：表结构变更审批、回滚方案
数据建模规范
星型模型
事实表（交易事实）
├── 维度：时间维度（日期）
├── 维度：产品维度（产品ID → 产品名称/品类/品牌）
├── 维度：用户维度（用户ID → 用户属性）
└── 维度：门店维度（门店ID → 门店属性）
分层 SQL 示例
-- ODS 层：原始数据
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;
-- DWD 层：清洗后明细
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}';
-- DWS 层：日汇总
CREATE TABLE dws_trade_daily (
dt STRING,
total_orders BIGINT,
total_amount DECIMAL(15,2),
paid_orders BIGINT,
paid_amount DECIMAL(15,2),
unique_users BIGINT
) STORED AS 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()
# 读取 MySQL
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))
# 写入 Delta Lake
df_clean.write \
.format("delta") \
.mode("overwrite") \
.partitionBy("dt") \
.option("replaceWhere", "dt >= '2024-01-01'") \
.save("/data/delta/dwd_trade_order")
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=3600,
poke_interval=300,
mode='reschedule',
)
extract_mysql = PythonOperator(
task_id='extract_mysql',
python_callable=extract_mysql_data,
op_kwargs={'table': 'orders', 'dt': '{{ ds }}'},
pool='extract_pool',
)
transform = PythonOperator(
task_id='transform',
python_callable=transform_data,
)
load_clickhouse = PythonOperator(
task_id='load_clickhouse',
python_callable=load_to_clickhouse,
)
quality_check = PythonOperator(
task_id='quality_check',
python_callable=run_quality_checks,
)
check_partition >> extract_mysql >> transform >> load_clickhouse >> quality_check
Flink SQL 实时聚合模板
-- Kafka 源表
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
FROM source_events
GROUP BY
TUMBLE(event_time, INTERVAL '5' MINUTE),
page_url;
-- ClickHouse 结果表
CREATE TABLE sink_pv_uv (
window_start TIMESTAMP(3),
window_end TIMESTAMP(3),
page_url STRING,
pv BIGINT,
uv BIGINT
) WITH (
'connector' = 'clickhouse',
'url' = 'clickhouse://host:8123',
'table-name' = 'dw_agg_pv_uv_5min',
'database' = 'analytics',
'sink.batch-size' = 1000,
'sink.flush-interval' = '1000'
);
-- 写入
INSERT INTO sink_pv_uv
SELECT window_start, window_end, page_url, pv, uv
FROM pv_uv_5min;
dbt 模型示例
-- models/staging/stg_orders.sql
-- 原始数据清洗层
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
-- models/marts/dim_product.sql
-- 维度表
SELECT
product_id,
product_name,
category,
brand,
price,
created_at
FROM {{ ref('stg_products') }}
WHERE is_active = true
-- models/marts/fct_orders.sql
-- 事实表，增量策略
{{ config(
materialized='incremental',
unique_key='order_id',
incremental_strategy='merge',
) }}
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('dim_product') }} p ON o.product_id = p.product_id
{% if is_incremental() %}
WHERE o.update_time > (SELECT max(update_time) FROM {{ 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 | 管道运维、排障、扩缩容指南 |