SuperAPI AI 工具集
合作ENAI 行业资讯收藏
Skill 广场agent-data-engineer

agent-data-engineer

数据与分析

查看项目

更新于 10-16 22:15

技能介绍

Data Engineer Agent

本条目介绍的是一个名为 agent-data-engineer 的专家子代理（subagent），定位为数据工程领域的专门代理，专注 ETL/ELT 管道设计、数据仓库架构、流处理、数据建模与数据质量保障，覆盖现代数据技术栈的多种工具。

主要能力

资料中列出的能力范围包括：

  • ETL/ELT 管道：Apache Airflow（DAGs、operators、sensors）、Dagster（assets、resources、IO managers）、Prefect（flows、tasks、deployments）、Luigi（任务依赖）、自定义 Python 管道，以及增量与全量刷新策略。
  • 数据仓库：BigQuery（分区、聚簇、物化视图）、Snowflake（warehouses、stages、streams、tasks）、Redshift（分布键、排序键、spectrum）、ClickHouse（实时分析）、DuckDB（嵌入式分析），以及数据湖模式（S3/GCS + catalog）。
  • 流处理：Apache Kafka（producers、consumers、Kafka Streams）、Apache Flink（有状态流处理）、AWS Kinesis（Data Streams、Firehose、Analytics）、Google Pub/Sub + Dataflow、Redis Streams、变更数据捕获（Debezium、CDC 模式）。
  • 数据建模：星型模式（事实表与维度表）、雪花模式、Data Vault（hubs、links、satellites）、用于分析的 One Big Table（OBT）、缓慢变化维度（SCD Type 1、2、3）、activity schema。
  • dbt：模型组织（staging、intermediate、marts）、增量模型、快照（SCD Type 2）、测试（schema、自定义、数据）、文档与血缘、宏与包。
  • 数据质量：Great Expectations（expectations、checkpoints）、dbt tests（unique、not_null、accepted_values、relationships）、数据契约与 schema 校验、异常检测、数据新鲜度监控、对账检查。
  • 数据治理：数据目录（DataHub、Amundsen、OpenMetadata）、列级血缘、PII 检测与脱敏、访问控制与 RBAC、数据保留策略。

适用场景

资料中列出的使用时机包括：为新数据平台设计 ETL/ELT 管道、搭建数据仓库（BigQuery、Snowflake、Redshift）、用 Kafka 实现实时流、为分析构建 dbt 模型、设计数据模型（星型模式、data vault）、搭建数据质量测试、实现 CDC 实时同步、优化数据仓库查询性能。

该代理的触发短语包括 ETL、pipeline、data warehouse、BigQuery、Snowflake、Redshift、Kafka、Airflow、dbt、streaming、data lake、data model、star schema。

使用方式与建议

资料中给出在该代理下处理数据工程任务时的五条指令：

  1. 理解数据流：在写代码前先映射源系统、转换与目标，先画管道。
  2. 尽可能选择 ELT 而非 ETL：先把原始数据加载进仓库，再用 dbt 转换，这样更灵活、更可审计。
  3. 幂等管道：每次管道运行无论执行多少次都应产生相同结果，使用 merge/upsert 模式而非仅插入。
  4. 在每个阶段测试数据质量：在摄取时、转换后、对外服务前进行校验，尽早发现问题。
  5. 为增量处理而设计：全量刷新不可扩展，从一开始就使用时间戳、水位线或 CDC 做增量加载。

示例代码与模式

资料中提供了多段示例代码，用于说明具体实践：

  • dbt 项目结构：展示 dbt_project 目录下 models（staging、intermediate、marts）、seeds、snapshots、macros、tests 的组织方式。
  • dbt staging 模型：对源表做重命名、金额换算、生成代理键等轻量清洗。
  • dbt 增量模型：通过 config 设置 materialized='incremental'、unique_key、merge 策略与 on_schema_change，并用 is_incremental() 结合 max(updated_at) 过滤增量数据。
  • dbt schema 测试：在 yml 中定义 unique、not_null、relationships、accepted_range、accepted_values 等测试。
  • Airflow DAG 模式：定义 default_args、DAG 调度（示例为每日 6 点 UTC）、以及 extract → load → dbt → data quality 的任务依赖链。
  • Kafka 消费者模式（Python）：使用 confluent_kafka 的 Consumer，设置 group.id、手动提交 offset、异常处理与死信队列（DLQ）。
  • 星型模式设计：给出 fct_orders 事实表、dim_customers 维度表（含 SCD Type 2 字段 valid_from/valid_to/is_current）与 dim_dates 日期维度表的建表语句。

需要注意，以上代码片段中部分内容未完整展示（如 dbt 宏 generate_surrogate_key 的实现、extract_stripe_to_gcs 与 run_great_expectations_suite 的具体实现、发送至 DLQ 的 send_to_dlq 实现等），这些部分的完整代码暂未提供。

数据管道检查清单

资料提供了一份清单：管道幂等（可安全重跑）、实现增量加载（非全量刷新）、带死信队列或重试的错误处理、摄取与转换阶段的数据质量测试、处理 schema 演进（新增列、类型变更）、管道失败的监控与告警、记录回填策略、按数据分级处理 PII、记录管道依赖（DAG 血缘）、用生产规模数据做性能测试。

关联技能

资料中列出的参考技能包括：data-science（统计分析与 ML 管道）、database-expert（数据库优化与 schema 设计）、agent-cloud-architect（数据平台基础设施）、devops-engineer（数据管道的 CI/CD）。

限制与未提供信息

资料未提供该子代理的安装命令、运行方式、收费或价格信息，也未提供其具体实现机制、支持平台、版本要求或效果数据，这些内容暂未提供。