data-engineering

name	data-engineering
description	数据工程（Airflow/Dagster/Kafka/Flink/dbt、数据管道、ETL、流处理、数据质量）。
license	MIT
user-invocable	false
disable-model-invocation	false
context	fork

数据工程域 · Data Engineering

域概览

数据工程域涵盖数据管道编排、流式处理、数据质量保障三大核心领域。

数据管道层                流处理层              质量保障层
├── Airflow (调度编排)    ├── Kafka Streams     ├── Great Expectations
├── Dagster (资产管理)    ├── Flink             ├── dbt
└── Prefect (现代工作流)  └── Spark Streaming   └── Soda Core

---

数据管道编排

框架对比

特性	Airflow	Dagster	Prefect
核心模型	DAG + Task	Asset + Op	Flow + Task
学习曲线	陡峭	中等	平缓
资产管理	无	原生支持	无
动态任务	支持	支持	支持
本地开发	复杂	简单	简单
社区生态	最大	成长中	成长中

Airflow 核心模式

• DAG 定义：with DAG(dag_id, schedule, default_args) as dag
• TaskFlow API：@task 装饰器，自动 XCom 传递
• 动态任务：@task + .expand() 实现 dynamic task mapping
• Operators：PythonOperator / BashOperator / SQL / HTTP / S3
• Sensors：FileSensor / HttpSensor / ExternalTaskSensor
• 重试策略：retries=3, retry_delay=timedelta(minutes=5), retry_exponential_backoff=True
• 失败回调：on_failure_callback 发送告警
• SLA 监控：sla=timedelta(hours=2) + sla_miss_callback

Dagster 核心模式

• Asset 定义：@asset(group_name, deps) 声明数据资产
• MaterializeResult：返回元数据（行数、预览等）
• Resources：ConfigurableResource 管理外部连接
• Jobs：define_asset_job(selection=AssetSelection.groups(...))
• Schedules：ScheduleDefinition(job, cron_schedule)
• Sensors：@sensor(job) 监听外部事件触发
• Partitions：DailyPartitionsDefinition 按日分区
• Asset Checks：@asset_check 验证数据新鲜度/质量

Prefect 核心模式

• Flow/Task：@flow + @task(retries=3, cache_key_fn=task_input_hash)
• 并发：ConcurrentTaskRunner + task.map(items)
• Deployments：Deployment.build_from_flow(schedule=CronSchedule(...))
• Blocks：Secret / JSON 管理配置和密钥

调度策略 Checklist

• Cron 表达式正确（0 2 * * * 日批 / */15 * * * * 实时）
• 事件驱动：文件到达 / S3 / API 触发
• 跨 DAG 依赖：ExternalTaskSensor / Asset deps
• 幂等性：UPSERT / 分区覆盖写入
• 增量处理：WHERE updated_at > last_run
• 数据血缘：Dagster 原生 / Airflow Lineage / dbt ref()

---

流式处理

框架对比

特性	Kafka Streams	Flink	Spark Streaming
部署模式	嵌入式(JVM)	独立集群	独立集群
状态管理	RocksDB	内存/RocksDB	内存
Exactly-Once	支持	支持	支持
窗口类型	丰富	最丰富	基础
学习曲线	平缓	陡峭	中等
Python API	kafka-python	PyFlink	PySpark

Kafka Streams 核心模式

• 拓扑构建：StreamsBuilder → stream() → filter/map/flatMap → to()
• 聚合：groupByKey().count() / .aggregate() / .reduce()
• Join：Stream-Stream（时间窗口）/ Stream-Table / Table-Table
• 状态存储：Stores.persistentKeyValueStore + Transformer
• Exactly-Once：PROCESSING_GUARANTEE_CONFIG = EXACTLY_ONCE_V2
• 性能调优：NUM_STREAM_THREADS=4 / CACHE_MAX_BYTES_BUFFERING / RocksDB 配置

Flink 核心模式

• DataStream API：env.addSource() → filter/map → addSink()
• 窗口类型：
  • 滚动窗口 TumblingProcessingTimeWindows.of(Time.minutes(5))
  • 滑动窗口 SlidingProcessingTimeWindows.of(size, slide)
  • 会话窗口 ProcessingTimeSessionWindows.withGap(gap)
  • 全局窗口 GlobalWindows.create() + 自定义 Trigger
• 窗口聚合：aggregate(AggregateFunction, WindowFunction) 增量+全窗口
• ProcessFunction：低级 API，访问时间戳、注册定时器
• 状态管理：ValueState / ListState / MapState + TTL 清理
• Checkpoint：env.enableCheckpointing(60000) + EXACTLY_ONCE
• Savepoint：flink run -s /path/to/savepoint
• 时间语义：Event Time + Watermark（forBoundedOutOfOrderness）
• 延迟数据：allowedLateness() + sideOutputLateData()
• 数据倾斜：添加随机前缀打散 key

流处理 Checklist

• 选择时间语义：Event Time vs Processing Time
• Watermark 策略：乱序容忍度设置
• 窗口类型匹配业务场景
• 状态 TTL 防止无限增长
• Checkpoint 间隔和超时配置
• Exactly-Once 语义端到端保证
• 背压监控和处理
• 并行度调优

---

数据质量

质量维度

完整性(非空) → 准确性(范围) → 一致性(关联) → 及时性(新鲜度) → 有效性(格式)

工具对比

工具	优势	适用场景
Great Expectations	丰富 Expectations、Data Docs	Python 生态、复杂验证
dbt	SQL 原生、血缘追踪	数据仓库、转换测试
Soda Core	简洁 YAML 配置	快速验证、CI/CD

Great Expectations 核心模式

• Data Context：gx.get_context() → 添加数据源 → 构建批次
• 常用 Expectations：
  • expect_table_row_count_to_be_between(min, max)
  • expect_column_values_to_not_be_null(column)
  • expect_column_values_to_be_unique(column)
  • expect_column_values_to_be_between(column, min, max)
  • expect_column_values_to_be_in_set(column, value_set)
  • expect_column_values_to_match_regex(column, regex)
• Checkpoints：批量运行验证 + 生成 Data Docs
• 自定义 Expectation：继承 ColumnMapExpectation

dbt 测试核心模式

• Schema 测试：unique / not_null / accepted_values / relationships
• Generic 测试：{% test name(model, column_name, params) %}
• Singular 测试：tests/ 目录下自定义 SQL，返回行 = 失败
• dbt_expectations 包：expect_column_mean_to_be_between / expect_row_values_to_have_recent_data
• 执行：dbt test / dbt test --select model / dbt test --store-failures
• 血缘：{{ ref('model') }} + {{ source('schema', 'table') }} → dbt docs generate

Soda Core 核心模式

checks for table_name:
  - row_count > 100
  - missing_count(column) = 0
  - duplicate_count(column) = 0
  - invalid_count(column) = 0:
      valid format: email
  - freshness(timestamp_col) < 1d

数据质量 Checklist

• 分层验证：源数据 → 转换后 → 目标数据
• 完整性：必需列非空、无空字符串
• 准确性：数值范围、格式正则、逻辑一致
• 一致性：跨表主键匹配、值一致
• 及时性：数据新鲜度 < 阈值
• 唯一性：主键/业务键无重复
• 质量指标：完整性/唯一性/有效性加权评分
• 告警：指标低于阈值自动通知（Slack/Email/PagerDuty）
• 持续监控：定时执行质量检查

---

最佳实践

实践	说明
幂等性设计	UPSERT / 分区覆盖，重跑不产生副作用
增量处理	基于时间戳/CDC 增量提取，减少全量扫描
数据血缘	dbt ref() / Dagster Asset deps 追踪上下游
分层验证	源→转换→目标每层都验证
监控告警	管道 SLA + 质量指标 + 延迟告警
状态管理	流处理状态 TTL + Checkpoint + Savepoint
容错设计	重试策略 + 死信队列 + 回滚方案

触发词

数据管道、Airflow、Dagster、Prefect、ETL、流处理、Kafka Streams、Flink、数据质量、Great Expectations、dbt、数据验证、数据血缘