senior-data-engineer

Senior Data Engineer 是一个面向生产环境的数据工程技能，目标在于构建可扩展且可靠的数据系统。资料将其定位为“生产级”（production-grade），强调可扩展性与可靠性。

用途与触发场景

该技能在出现以下类型需求时被激活：

• 管道设计：例如“为……设计数据管道”“构建 ETL/ELT 流程”“如何从……摄入数据”“从……设置数据抽取”。
• 架构决策：例如“应该用批处理还是流处理”“Lambda 与 Kappa 架构”“如何处理迟到数据”“设计数据湖仓（lakehouse）”。
• 数据建模：例如“创建维度模型”“星型模型与雪花模型”“实现缓慢变化维度”“设计 data vault”。
• 数据质量：例如“为……添加数据校验”“设置数据质量检查”“监控数据新鲜度”“实现数据契约（data contracts）”。
• 性能优化：例如“优化这个 Spark 作业”“查询很慢”“减少管道执行时间”“调优 Airflow DAG”。

主要能力与工作流

资料给出三条主要工作流。

工作流一：构建批量 ETL 管道

场景为从 PostgreSQL 抽取数据、用 dbt 转换、加载到 Snowflake。步骤包括：

1. 定义源 schema：通过 information_schema.columns 记录源表结构。
2. 生成抽取配置：调用 pipeline_orchestrator.py generate，指定 airflow 类型、postgres 源、增量模式、以 updated_at 作为水位线，输出到 dags/extract_source.py。
3. 创建 dbt 模型：编写 staging 层（stg_orders）与 marts 层（fct_orders），后者配置为增量物化、以 order_id 为唯一键、按 order_date 聚簇。
4. 配置数据质量测试：在 schema.yml 中对 order_id 做 unique 与 not_null，对 total_amount 做非空与取值区间校验，对 order_date 做非空与新鲜度（recency）检查。
5. 创建 Airflow DAG：daily_etl_pipeline 每日 5 点调度，任务链为 extract >> transform >> test >> notify，含重试与失败邮件告警配置。
6. 验证管道：本地运行 dbt run 与 dbt test，并用 data_quality_validator.py 校验并输出质量报告。

工作流二：实现实时流处理

场景为从 Kafka 流式读取事件、用 Spark Streaming 处理、写入数据湖。步骤包括：

1. 定义事件 schema：JSON Schema 描述 UserEvent，必填 event_id、user_id、event_type、timestamp，event_type 枚举含 page_view、click、purchase。
2. 创建 Kafka 主题：user-events，12 个分区、3 副本，配置保留与清理策略，并 describe 验证。
3. 实现 Spark Streaming 作业：读取 Kafka、解析 JSON、按 5 分钟窗口与事件类型聚合，带 10 分钟 watermark，结果以 append 模式写入 Delta Lake，处理间隔 1 分钟。
4. 处理迟到数据与错误：使用 foreachBatch，将有效记录写入数据湖，将缺少 event_id 的无效记录写入死信队列（DLQ）。
5. 监控流健康：使用 Prometheus 客户端定义已处理记录数、处理延迟、批次耗时等指标。

工作流三：数据质量框架配置

场景为用 Great Expectations 实施全面的数据质量监控。步骤包括：

1. 初始化 Great Expectations：安装后执行 init 与 datasource new。
2. 创建期望套件：涵盖列顺序、非空、唯一性、取值范围、枚举集合、新鲜度与引用完整性等期望，并保存套件。
3. 用 dbt 创建数据质量检查：行数对比、新鲜度、唯一性、非空、引用关系与数值区间等测试。
4. 实现数据契约：YAML 定义 schema、SLA（新鲜度最大延迟小时数、完整性最低百分比、重复容忍度）与消费者（分析团队、ML 团队）。
5. 设置质量监控看板：生成包含行数、新鲜度、空值率等检查项的质量报告。

使用方式

该技能提供若干脚本命令作为核心工具入口：

• 生成管道编排配置：python scripts/pipeline_orchestrator.py generate --type airflow --source postgres --destination snowflake --schedule "0 5 * * *"
• 校验数据质量：python scripts/data_quality_validator.py validate --input data/sales.parquet --schema schemas/sales.json --checks freshness,completeness,uniqueness
• 优化 ETL 性能：python scripts/etl_performance_optimizer.py analyze --query queries/daily_aggregation.sql --engine spark --recommend 资料还提及技术栈（Tech Stack）、参考文档（Reference Documentation）与故障排除（Troubleshooting）章节标题，但具体内容暂未提供。

适用场景与限制

适用场景包括批处理与流处理管道建设、维度建模、数据质量与数据契约落地、Spark 与 Airflow 性能调优等数据工程任务。限制方面：资料未说明该技能的运行环境依赖、支持的版本范围、收费方式或价格信息，暂未提供；其触发词、脚本与代码示例虽已给出，但未列出性能基准或效果保证。