data-engineering

name data-engineering description 数据工程。Airflow、Dagster、Kafka Streams、Flink、dbt、数据管道、流处理、数据质量。当用户提到数据管道、ETL、流处理、数据质量时路由到此。 license MIT user-invocable false disable-model-invocation false 数据工程域 · 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、数据验证、数据血缘