mcpField
D
data-engineering

data-engineering

github.com/travisjneuman数据与数据库106100 次安装今天更新
代码仓库

关于 data-engineering

用于设计数据管道、转换、流处理系统、仓库架构和数据质量检查的数据工程技能。介绍 Airflow、Dagster、dbt、Kafka 和 Great Expectations 等工具相关模式与示例。

主要功能

  1. 01ETL 与 ELT 架构模式
  2. 02Airflow 和 Dagster 编排
  3. 03dbt 转换示例
  4. 04Kafka 流处理
  5. 05仓库架构与质量实践
  6. 06增量及幂等管道模式

适用场景

  1. 01选择 ETL 或 ELT 方案
  2. 02规划编排式数据管道
  3. 03设计仓库表与质量检查
  4. 04制定增量处理和回填策略

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/travisjneuman/.claude --skill data-engineering

安装后在 Claude Code 中输入/data-engineering确认已加载。

详细介绍

数据工程

流水线架构

ETL 与 ELT
模式适用场景工具
ETL加载前进行转换，数据质量至关重要Airflow + 自定义工具、Spark
ELT原始数据 → 数据仓库 → 原地转换Fivetran + dbt、Airbyte + dbt
编排

Apache Airflow：

from airflow.decorators import dag, task
from datetime import datetime

@dag(schedule="@daily", start_date=datetime(2024, 1, 1), catchup=False)
def my_pipeline():
    @task()
    def extract() -> dict:
        return {"data": "extracted"}

    @task()
    def transform(data: dict) -> dict:
        return {"transformed": True}

    @task()
    def load(data: dict):
        # Load to warehouse
        pass

    raw = extract()
    transformed = transform(raw)
    load(transformed)

my_pipeline()

Dagster（推荐用于新项目）：

from dagster import asset, Definitions

@asset
def raw_users():
    return extract_from_source()

@asset
def cleaned_users(raw_users):
    return clean_and_validate(raw_users)

dbt 转换

-- models/marts/dim_customers.sql
{{ config(materialized='table', schema='marts') }}

WITH source AS (
    SELECT * FROM {{ ref('stg_customers') }}
),
orders AS (
    SELECT customer_id, COUNT(*) as order_count, SUM(amount) as total_spent
    FROM {{ ref('stg_orders') }}
    GROUP BY customer_id
)
SELECT
    s.customer_id,
    s.name,
    s.email,
    COALESCE(o.order_count, 0) as lifetime_orders,
    COALESCE(o.total_spent, 0) as lifetime_value
FROM source s
LEFT JOIN orders o ON s.customer_id = o.customer_id

流处理

Apache Kafka：

from confluent_kafka import Producer, Consumer

# Producer
producer = Producer({'bootstrap.servers': 'localhost:9092'})
producer.produce('events', key='user_123', value=json.dumps(event))
producer.flush()

# Consumer
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest'
})
consumer.subscribe(['events'])

数据仓库模式设计

星型模式
  • 事实表： 可度量的事件（订单、点击、交易）
  • 维度表： 描述性上下文（客户、产品、日期）
  • 缓慢变化维度： 类型 1（覆盖）、类型 2（版本化行）、类型 3（前一列）
数据质量
  • Great Expectations： 模式验证、统计测试、自定义期望
  • dbt 测试： not_null、unique、accepted_values、relationships、自定义 SQL 测试
  • 数据契约： 模式演进策略、向后兼容要求

关键模式

  • 幂等流水线： 相同输入始终产生相同输出，可安全重跑
  • 增量模型： 仅处理新增或变更的数据，使用 updated_at 水位标记
  • 死信队列： 将失败记录转移以供检查，而不阻塞流水线
  • 回填策略： 按时间分区的表支持定向历史数据重新处理

评论

支持 Markdown