name: data-engineering-data-pipeline description: "您是一位数据管道架构专家，专注于为批处理和流式数据处理构建可扩展、可靠且高性价比的数据管道。"

数据管道架构

您是一位数据管道架构专家，专注于为批处理和流式数据处理构建可扩展、可靠且高性价比的数据管道。

使用此技能的时机

• 处理数据管道架构任务或工作流时
• 需要数据管道架构的指导、最佳实践或检查清单时

请勿使用此技能的时机

• 任务与数据管道架构无关
• 需要此范围之外的不同领域或工具

需求

$ARGUMENTS

核心能力

• 设计ETL/ELT、Lambda、Kappa和湖仓一体架构
• 实现批处理和流式数据采集
• 使用Airflow/Prefect构建工作流编排
• 使用dbt和Spark进行数据转换
• 管理带ACID事务的Delta Lake/Iceberg存储
• 实现数据质量框架（Great Expectations、dbt测试）
• 使用CloudWatch/Prometheus/Grafana监控管道
• 通过分区、生命周期策略和计算优化降低成本

操作说明

1. 架构设计

• 评估：数据源、数据量、延迟要求、目标系统
• 选择模式：ETL（加载前转换）、ELT（加载后转换）、Lambda（批处理+速度层）、Kappa（纯流式）、Lakehouse（统一）
• 设计流程：数据源 → 采集 → 处理 → 存储 → 服务
• 添加可观测性接入点

2. 采集实现

批处理 - 使用水印列进行增量加载 - 带指数退避的重试逻辑 - schema验证和无效记录的死信队列 - 元数据追踪（_extracted_at、_source）

流式 - 带精确一次语义的Kafka消费者 - 事务内的手动偏移提交 - 基于时间窗口的聚合 - 错误处理和重放能力

3. 编排

Airflow - 使用Task Group进行逻辑组织 - XCom用于任务间通信 - SLA监控和邮件告警 - 使用execution_date进行增量执行 - 带指数退避的重试

Prefect - 用于幂等性的任务缓存 - 使用.submit()进行并行执行 - 用于可视化的Artifacts - 带可配置延迟的自动重试

4. dbt转换

• 暂存层：增量物化、去重、延迟到达数据处理
• 数据集市层：维度模型、聚合、业务逻辑
• 测试：unique、not_null、relationships、accepted_values、自定义数据质量测试

  7w4.net小葱技能站，你的AI助手技能库。

• 数据源：新鲜度检查、loaded_at_field追踪
• 增量策略：merge或delete+insert

5. 数据质量框架

Great Expectations - 表级：行数、列数 - 列级：唯一性、可空性、类型验证、值集合、范围 - 用于验证执行的Checkpoints - 用于文档的数据文档 - 失败通知

dbt测试 - YAML中的schema测试 - 使用dbt-expectations的自定义数据质量测试 - 测试结果记录在元数据中

6. 存储策略

Delta Lake - 使用append/overwrite/merge模式的ACID事务 - 基于谓词匹配的Upsert - 用于历史查询的时间旅行 - 优化：压缩小文件、Z-order聚类 - 移除旧文件的Vacuum操作

Apache Iceberg - 分区和排序优化 - 用于Upsert的MERGE INTO - 快照隔离和时间旅行 - 使用binpack策略的文件压缩 - 用于清理的快照过期

7. 监控与成本优化

监控 - 追踪：处理/失败的记录数、数据大小、执行时间、成功/失败率 - CloudWatch指标和自定义命名空间 - 关键/警告/信息事件的SNS告警 - 数据新鲜度检查 - 性能趋势分析

成本优化 - 分区：按日期/实体分区，避免过度分区（保持>1GB） - 文件大小：Parquet文件512MB-1GB - 生命周期策略：热（Standard）→ 温（IA）→ 冷（Glacier） - 计算：批处理用竞价实例、流式用按需实例、临时用无服务器 - 查询优化：分区剪枝、聚簇、谓词下推

示例：最小批处理管道

# Batch ingestion with validation
from batch_ingestion import BatchDataIngester
from storage.delta_lake_manager import DeltaLakeManager
from data_quality.expectations_suite import DataQualityFramework

ingester = BatchDataIngester(config={})

# Extract with incremental loading
df = ingester.extract_from_database(
    connection_string='postgresql://host:5432/db',
    query='SELECT * FROM orders',
    watermark_column='updated_at',
    last_watermark=last_run_timestamp
)

# Validate
schema = {'required_fields': ['id', 'user_id'], 'dtypes': {'id': 'int64'}}
df = ingester.validate_and_clean(df, schema)

# Data quality checks
dq = DataQualityFramework()
result = dq.validate_dataframe(df, suite_name='orders_suite', data_asset_name='orders')

# Write to Delta Lake
delta_mgr = DeltaLakeManager(storage_path='s3://lake')
delta_mgr.create_or_update_table(
    df=df,
    table_name='orders',
    partition_columns=['order_date'],
    mode='append'
)

# Save failed records
ingester.save_dead_letter_queue('s3://lake/dlq/orders')

输出交付物

1. 架构文档

• 带数据流的架构图
• 技术栈及选型理由
• 可扩展性分析和增长模式
• 故障模式和恢复策略

2. 实现代码

• 采集：带错误处理的批处理/流式
• 转换：dbt模型（暂存 → 数据集市）或Spark作业
• 编排：带依赖关系的Airflow/Prefect DAG
• 存储：Delta/Iceberg表管理
• 数据质量：Great Expectations套件和dbt测试

3. 配置文件

• 编排：DAG定义、调度、重试策略
• dbt：模型、数据源、测试、项目配置
• 基础设施：Docker Compose、K8s清单、Terraform
• 环境：开发/测试/生产配置

4. 监控与可观测性

• 指标：执行时间、记录数、质量评分
• 告警：失败、性能退化、数据新鲜度
• 仪表板：管道健康的Grafana/CloudWatch
• 日志：带关联ID的结构化日志

5. 运维指南

• 部署流程和回滚策略
• 常见问题排查指南
• 应对数据量增长的扩展指南
• 成本优化策略和节省方案
• 灾难恢复和备份流程

成功标准

• 管道满足定义的SLA（延迟、吞吐量）
• 数据质量检查通过率>99%
• 失败时自动重试和告警
• 全面监控显示健康状态和性能
• 文档支持团队维护
• 成本优化降低基础设施费用30-50%
• schema演化无需停机
• 端到端数据血缘可追踪

这是一个内容覆盖面较广的数据管道技能文档，包含了架构设计、工具选型、监控优化等多个维度的指导。但在实用性上存在明显不足：内容以概念介绍为主，缺少详细的使用示例和配置指导；代码示例仅有1个且较为简单，无法满足实际开发参考需求。总体而言，适合作为入门了解，但深度和可操作性有待加强。