数据工程资源专题

Data Engineering Resources

---

🧭 学习路径

阶段	核心技能	工具
🟢 基础	SQL 精通、Python 数据处理	PostgreSQL、Pandas
🟡 管道	ETL/ELT、工作流调度	Airflow、dbt、Prefect
🟠 大数据	分布式计算、列式存储	Spark、Parquet、Delta Lake
🔴 流处理	实时数据管道	Kafka、Flink、RisingWave
⚫ 平台	数据湖、数据网格、MLOps	Databricks、Snowflake

数据是决策的原料。先保证数据质量 → 再谈分析深度。

---

📚 概述

数据工程是设计、构建和维护数据基础设施的学科，确保数据能够被有效采集、存储、处理和访问。数据工程师为数据科学家和分析师提供可靠的数据管道和平台。

核心职责: - 数据管道设计与构建 - 数据仓库与数据湖架构 - ETL/ELT 流程开发 - 数据质量管理 - 实时数据处理 - 数据治理与安全

---

🗺️ 知识地图

graph TB
    A[数据工程] --> B[数据采集]
    A --> C[数据存储]
    A --> D[数据处理]
    A --> E[数据服务]

    B --> B1[批处理采集]
    B --> B2[流式采集]
    B --> B3[API 集成]
    B --> B4[CDC 变更捕获]

    C --> C1[关系数据库]
    C --> C2[NoSQL]
    C --> C3[数据仓库]
    C --> C4[数据湖]

    D --> D1[批处理]
    D --> D2[流处理]
    D --> D3[数据转换]
    D --> D4[数据质量]

    E --> E1[数据 API]
    E --> E2[BI 工具]
    E --> E3[数据目录]
    E --> E4[权限管理]

---

📖 经典书籍

入门级

书名	作者	年份	难度	特点
《Fundamentals of Data Engineering》	Joe Reis & Matt Housley	2022	⭐⭐⭐	全面概览，最佳实践
《数据工程师手册》	王嘉诚	2021	⭐⭐⭐	中文实战，案例丰富
《Data Pipelines Pocket Reference》	James Densmore	2021	⭐⭐⭐	精简实用

进阶级

书名	作者	年份	难度	特点
《Designing Data-Intensive Applications》	Martin Kleppmann	2017	⭐⭐⭐⭐	DDIA，数据系统圣经
《The Data Warehouse Toolkit》	Ralph Kimball	2013	⭐⭐⭐⭐	维度建模经典
《Streaming Systems》	Akidau et al.	2018	⭐⭐⭐⭐	流处理权威指南

高级级

书名	作者	年份	难度	特点
《Database Internals》	Alex Petrov	2019	⭐⭐⭐⭐⭐	数据库内部原理
《Data Mesh》	Zhamak Dehghani	2022	⭐⭐⭐⭐⭐	下一代数据架构
《Building Evolutionary Architectures》	Ford et al.	2017	⭐⭐⭐⭐⭐	演进式架构

---

🎓 在线课程

免费课程

课程	平台	机构	时长	链接
Data Engineering on Google Cloud	Coursera	Google Cloud	6 门课	链接
Big Data Specialization	Coursera	UC San Diego	8 门课	链接
大数据技术	中国大学 MOOC	北京大学	12 周	链接

实战课程

课程	平台	特点	链接
Data Engineer Nanodegree	Udacity	项目驱动，导师指导	链接
Data Engineering with Python	DataCamp	交互式学习	链接
Apache Spark 实战	Udemy	Spark 深度讲解	链接

---

🔧 技术栈总览

数据采集

工具	类型	特点	链接
Airbyte	ELT 平台	开源，300+ 连接器	GitHub
Fivetran	ELT 服务	托管服务，自动化	官网
Debezium	CDC	实时变更捕获	GitHub
Logstash	日志采集	ELK 栈组件	GitHub
Flume	日志采集	Apache 项目	官网

数据存储

工具	类型	特点	链接
PostgreSQL	关系数据库	ACID，扩展性强	官网
MongoDB	文档数据库	灵活 Schema	GitHub
Snowflake	云数据仓库	弹性扩展，分离存储计算	官网
Delta Lake	数据湖格式	ACID 事务，版本控制	GitHub
Apache Iceberg	数据湖格式	高性能，分区演进	官网

数据处理

工具	类型	特点	链接
Apache Spark	批/流处理	内存计算，生态完善	GitHub
Apache Flink	流处理	低延迟，精确一次	GitHub
dbt	数据转换	SQL 优先，版本控制	GitHub
Apache Beam	统一批流	多引擎支持	官网

工作流编排

工具	类型	特点	链接
Apache Airflow	工作流调度	Python DAG，生态丰富	GitHub
Prefect	工作流编排	现代 API，易调试	GitHub
Dagster	数据编排	数据感知，开发体验好	GitHub
Kestra	工作流自动化	YAML 定义，UI 友好	GitHub

---

📊 核心技术详解

1. ETL 管道设计

# 使用 Prefect 构建 ETL 管道
from prefect import task, Flow, Parameter
from prefect.executors import LocalDaskExecutor
import pandas as pd
import sqlalchemy as sa

@task
def extract_data(source_url: str) -> pd.DataFrame:
    """从数据源提取数据"""
    # 示例：从 CSV 或 API 提取
    if source_url.endswith('.csv'):
        df = pd.read_csv(source_url)
    elif source_url.endswith('.json'):
        df = pd.read_json(source_url)
    else:
        # 数据库连接
        engine = sa.create_engine(source_url)
        df = pd.read_sql("SELECT * FROM source_table", engine)

    print(f"提取数据：{len(df)} 行，{len(df.columns)} 列")
    return df

@task
def transform_data(df: pd.DataFrame) -> pd.DataFrame:
    """数据转换"""
    # 数据清洗
    df = df.drop_duplicates()
    df = df.dropna(subset=['id', 'created_at'])

    # 数据类型转换
    df['created_at'] = pd.to_datetime(df['created_at'])
    df['amount'] = df['amount'].astype(float)

    # 业务逻辑
    df['profit'] = df['revenue'] - df['cost']
    df['profit_margin'] = df['profit'] / df['revenue']

    # 数据验证
    assert (df['amount'] >= 0).all(), "金额不能为负数"
    assert (df['profit_margin'] <= 1).all(), "利润率不能超过 100%"

    print(f"转换后数据：{len(df)} 行")
    return df

@task
def load_data(df: pd.DataFrame, target_url: str, table_name: str):
    """加载数据到目标"""
    engine = sa.create_engine(target_url)

    # 增量加载或全量替换
    df.to_sql(
        table_name,
        engine,
        if_exists='replace',  # 或 'append'
        index=False,
        method='multi',
        chunksize=1000
    )

    print(f"数据已加载到 {table_name}")

@task
def send_notification(success: bool):
    """发送通知"""
    if success:
        print("✅ 数据管道执行成功")
        # 可集成 Slack、邮件等通知
    else:
        print("❌ 数据管道执行失败")

# 定义工作流
with Flow("ETL Pipeline", executor=LocalDaskExecutor()) as flow:
    source_url = Parameter("source_url", default="postgresql://user:pass@host/db")
    target_url = Parameter("target_url", default="postgresql://user:pass@host/warehouse")
    table_name = Parameter("table_name", default="fact_sales")

    raw_data = extract_data(source_url)
    cleaned_data = transform_data(raw_data)
    load_result = load_data(cleaned_data, target_url, table_name)

    # 成功通知
    send_notification(True)

# 运行管道
if __name__ == "__main__":
    flow.register(project_name="Data Engineering")
    # flow.run()

---

2. 数据仓库建模

# Kimball 维度建模示例
# 事实表 + 维度表设计

"""
-- 事实表：销售事实
CREATE TABLE fact_sales (
    sale_id BIGINT PRIMARY KEY,
    date_id INTEGER NOT NULL,
    customer_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    store_id INTEGER NOT NULL,

    -- 度量
    quantity INTEGER NOT NULL,
    unit_price DECIMAL(10, 2) NOT NULL,
    total_amount DECIMAL(12, 2) NOT NULL,
    cost DECIMAL(12, 2) NOT NULL,
    profit DECIMAL(12, 2) GENERATED ALWAYS AS (total_amount - cost) STORED,

    -- 外键
    FOREIGN KEY (date_id) REFERENCES dim_date(date_id),
    FOREIGN KEY (customer_id) REFERENCES dim_customer(customer_id),
    FOREIGN KEY (product_id) REFERENCES dim_product(product_id),
    FOREIGN KEY (store_id) REFERENCES dim_store(store_id)
);

-- 维度表：日期维度
CREATE TABLE dim_date (
    date_id INTEGER PRIMARY KEY,
    full_date DATE NOT NULL,
    year INTEGER,
    quarter INTEGER,
    month INTEGER,
    month_name VARCHAR(20),
    day_of_month INTEGER,
    day_of_week INTEGER,
    day_name VARCHAR(20),
    is_weekend BOOLEAN,
    is_holiday BOOLEAN
);

-- 维度表：客户维度 (SCD Type 2)
CREATE TABLE dim_customer (
    customer_key SERIAL PRIMARY KEY,
    customer_id INTEGER NOT NULL,

    -- 属性
    customer_name VARCHAR(100),
    email VARCHAR(100),
    phone VARCHAR(20),
    address VARCHAR(200),
    city VARCHAR(50),
    country VARCHAR(50),
    segment VARCHAR(50),

    -- SCD Type 2 元数据
    effective_date DATE NOT NULL,
    expiration_date DATE,
    is_current BOOLEAN DEFAULT TRUE,
    version INTEGER DEFAULT 1,

    UNIQUE (customer_id, effective_date)
);

-- 维度表：产品维度
CREATE TABLE dim_product (
    product_key SERIAL PRIMARY KEY,
    product_id INTEGER NOT NULL,
    product_name VARCHAR(200),
    category VARCHAR(50),
    subcategory VARCHAR(50),
    brand VARCHAR(50),
    sku VARCHAR(50),
    cost_price DECIMAL(10, 2),
    list_price DECIMAL(10, 2),
    is_active BOOLEAN DEFAULT TRUE
);
"""

# 使用 dbt 进行数据转换
# models/fact_sales.sql
"""
{{
    config(
        materialized='incremental',
        unique_key='sale_id',
        incremental_strategy='merge'
    )
}}

WITH source AS (
    SELECT * FROM {{ source('raw', 'sales') }}
),

dim_date_lookup AS (
    SELECT * FROM {{ ref('dim_date') }}
),

dim_customer_lookup AS (
    SELECT * FROM {{ ref('dim_customer') }} WHERE is_current = TRUE
),

final AS (
    SELECT
        {{ dbt_utils.generate_surrogate_key(['s.sale_id']) }} AS sale_key,
        d.date_id,
        c.customer_key,
        p.product_key,
        s.store_id,
        s.quantity,
        s.unit_price,
        s.quantity * s.unit_price AS total_amount,
        s.cost,
        (s.quantity * s.unit_price) - s.cost AS profit,
        CURRENT_TIMESTAMP AS created_at
    FROM source s
    JOIN dim_date_lookup d ON DATE(s.sale_date) = d.full_date
    JOIN dim_customer_lookup c ON s.customer_id = c.customer_id
    JOIN {{ ref('dim_product') }} p ON s.product_id = p.product_id

    {% if is_incremental() %}
        WHERE s.sale_date > (SELECT MAX(full_date) FROM {{ ref('dim_date') }})
    {% endif %}
)

SELECT * FROM final
"""

---

3. 实时数据处理

# Apache Flink 流处理示例
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings

def flink_streaming_pipeline():
    """Flink 实时数据处理"""

    # 创建执行环境
    env = StreamExecutionEnvironment.get_execution_environment()
    settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
    table_env = StreamTableEnvironment.create(env, environment_settings=settings)

    # 定义源表 (Kafka)
    table_env.execute_sql("""
        CREATE TABLE source_orders (
            order_id BIGINT,
            user_id BIGINT,
            amount DOUBLE,
            status STRING,
            order_time TIMESTAMP(3),
            WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
        ) WITH (
            'connector' = 'kafka',
            'topic' = 'orders',
            'properties.bootstrap.servers' = 'localhost:9092',
            'properties.group.id' = 'order-analysis',
            'format' = 'json',
            'scan.startup.mode' = 'latest-offset'
        )
    """)

    # 定义结果表
    table_env.execute_sql("""
        CREATE TABLE hourly_stats (
            window_start TIMESTAMP(3),
            window_end TIMESTAMP(3),
            total_orders BIGINT,
            total_amount DOUBLE,
            avg_amount DOUBLE,
            unique_users BIGINT
        ) WITH (
            'connector' = 'print'
        )
    """)

    # 实时聚合查询
    table_env.execute_sql("""
        INSERT INTO hourly_stats
        SELECT
            TUMBLE_START(order_time, INTERVAL '1' HOUR) AS window_start,
            TUMBLE_END(order_time, INTERVAL '1' HOUR) AS window_end,
            COUNT(*) AS total_orders,
            SUM(amount) AS total_amount,
            AVG(amount) AS avg_amount,
            COUNT(DISTINCT user_id) AS unique_users
        FROM source_orders
        WHERE status = 'completed'
        GROUP BY TUMBLE(order_time, INTERVAL '1' HOUR)
    """)

    # 实时异常检测 (CEP)
    table_env.execute_sql("""
        CREATE TABLE suspicious_orders (
            user_id BIGINT,
            order_count BIGINT,
            total_amount DOUBLE,
            window_start TIMESTAMP(3),
            window_end TIMESTAMP(3)
        ) WITH (
            'connector' = 'kafka',
            'topic' = 'suspicious-orders',
            'properties.bootstrap.servers' = 'localhost:9092',
            'format' = 'json'
        )
    """)

    # 检测 5 分钟内下单超过 10 次且总金额超过 10000 的用户
    table_env.execute_sql("""
        INSERT INTO suspicious_orders
        SELECT
            user_id,
            COUNT(*) AS order_count,
            SUM(amount) AS total_amount,
            TUMBLE_START(order_time, INTERVAL '5' MINUTE),
            TUMBLE_END(order_time, INTERVAL '5' MINUTE)
        FROM source_orders
        GROUP BY user_id, TUMBLE(order_time, INTERVAL '5' MINUTE)
        HAVING COUNT(*) > 10 AND SUM(amount) > 10000
    """)

# 运行流处理
if __name__ == "__main__":
    flink_streaming_pipeline()

---

4. 数据质量管理

# 使用 Great Expectations 进行数据质量检查
import great_expectations as ge
from great_expectations.core.expectation_configuration import ExpectationConfiguration
from great_expectations.data_context import DataContext

def setup_data_quality_framework():
    """设置数据质量框架"""

    # 初始化 DataContext
    context = DataContext()

    # 创建批次
    batch = context.get_batch(
        batch_kwargs={"table": "fact_sales", "schema": "public"},
        expectation_suite_name="sales_quality_suite"
    )

    # 定义期望
    expectations = [
        # 非空检查
        ExpectationConfiguration(
            expectation_type="expect_column_values_to_not_be_null",
            kwargs={"column": "sale_id"}
        ),
        ExpectationConfiguration(
            expectation_type="expect_column_values_to_not_be_null",
            kwargs={"column": "total_amount"}
        ),

        # 值域检查
        ExpectationConfiguration(
            expectation_type="expect_column_values_to_be_between",
            kwargs={"column": "quantity", "min_value": 0}
        ),
        ExpectationConfiguration(
            expectation_type="expect_column_values_to_be_between",
            kwargs={"column": "unit_price", "min_value": 0}
        ),

        # 唯一性检查
        ExpectationConfiguration(
            expectation_type="expect_column_values_to_be_unique",
            kwargs={"column": "sale_id"}
        ),

        # 分布检查
        ExpectationConfiguration(
            expectation_type="expect_column_mean_to_be_between",
            kwargs={"column": "profit_margin", "min_value": 0, "max_value": 1}
        ),

        # 完整性检查
        ExpectationConfiguration(
            expectation_type="expect_table_row_count_to_be_between",
            kwargs={"min_value": 1000}
        ),

        # 业务规则检查
        ExpectationConfiguration(
            expectation_type="expect_compound_columns_to_be_unique",
            kwargs={"column_list": ["order_id", "product_id"]}
        )
    ]

    # 添加期望到批次
    for exp in expectations:
        batch.add_expectation(exp)

    # 验证数据
    results = batch.validate()

    # 生成报告
    if results.success:
        print("✅ 数据质量检查通过")
    else:
        print("❌ 数据质量检查失败")
        for result in results.results:
            if not result.success:
                print(f"  - {result.expectation_config.expectation_type}: {result.expectation_config.kwargs}")

    return results

# 使用 dbt 测试
# models/schema.yml
"""
version: 2

models:
  - name: fact_sales
    columns:
      - name: sale_id
        tests:
          - unique
          - not_null

      - name: total_amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

      - name: quantity
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              integer: true

      - name: profit_margin
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1

      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customer')
              field: customer_key
"""

---

5. 数据目录与治理

# 使用 Apache Atlas 进行数据治理
from atlasclient.client import Atlas

def setup_data_governance():
    """设置数据治理框架"""

    # 连接 Atlas
    atlas = Atlas(
        host='localhost',
        port=21000,
        username='admin',
        password='admin'
    )

    # 创建数据资产
    table_entity = {
        "typeName": "DataSet",
        "attributes": {
            "qualifiedName": "warehouse.sales.fact_sales",
            "name": "fact_sales",
            "description": "销售事实表，包含所有交易记录",
            "owner": "data-engineering-team",
            "schema": [
                {"name": "sale_id", "type": "BIGINT", "description": "销售 ID"},
                {"name": "date_id", "type": "INTEGER", "description": "日期 ID"},
                {"name": "customer_id", "type": "INTEGER", "description": "客户 ID"},
                {"name": "total_amount", "type": "DECIMAL", "description": "总金额"},
                {"name": "profit", "type": "DECIMAL", "description": "利润"}
            ],
            "classification": ["PII", "BUSINESS_CRITICAL"],
            "tags": ["sales", "fact-table", "daily-refresh"]
        }
    }

    # 创建血缘关系
    lineage_entity = {
        "typeName": "Process",
        "attributes": {
            "qualifiedName": "etl.sales_pipeline",
            "name": "Sales ETL Pipeline",
            "inputs": [
                {"typeName": "DataSet", "uniqueAttributes": {"qualifiedName": "raw.sales"}}
            ],
            "outputs": [
                {"typeName": "DataSet", "uniqueAttributes": {"qualifiedName": "warehouse.sales.fact_sales"}}
            ],
            "description": "每日销售数据 ETL 流程"
        }
    }

    # 上传实体
    atlas.entities.create(entities=[table_entity, lineage_entity])

    print("✅ 数据资产和血缘关系已创建")

# 数据发现与搜索
def search_data_assets(search_query: str):
    """搜索数据资产"""
    atlas = Atlas(host='localhost', port=21000, username='admin', password='admin')

    results = atlas.search_dsl(query=search_query)

    print(f"找到 {len(results)} 个数据资产:")
    for entity in results:
        print(f"  - {entity.name} ({entity.typeName})")
        print(f"    描述：{entity.description}")
        print(f"    标签：{', '.join(entity.tags)}")

    return results

---

📝 最佳实践

数据管道设计原则

data_pipeline_principles = {
    '可靠性': [
        '幂等性：多次执行产生相同结果',
        '容错性：失败后自动重试',
        '监控：关键指标告警',
        '审计：操作日志记录'
    ],
    '可扩展性': [
        '水平扩展：支持分布式处理',
        '增量处理：仅处理变化数据',
        '分区策略：按时间/业务分区',
        '资源隔离：关键任务优先'
    ],
    '可维护性': [
        '代码版本控制：Git 管理',
        '配置外部化：环境分离',
        '文档完善：数据字典、血缘',
        '测试覆盖：单元测试、集成测试'
    ],
    '数据质量': [
        '入口校验：Schema 验证',
        '过程监控：质量指标',
        '出口检查：完整性验证',
        '异常处理：死信队列'
    ]
}

for principle, practices in data_pipeline_principles.items():
    print(f"\n{principle}:")
    for practice in practices:
        print(f"  • {practice}")

---

性能优化技巧

# 1. 批处理优化
batch_optimization_tips = """
- 批量写入：使用 batch insert 而非单条插入
- 并行处理：多进程/多线程并行
- 分区剪枝：只读取需要的分区
- 列式存储：使用 Parquet/ORC 格式
- 数据压缩：Snappy/Zstd 压缩
"""

# 2. SQL 优化
sql_optimization_tips = """
- 避免 SELECT *：只选择需要的列
- 使用分区字段过滤
- 合理使用索引
- 避免笛卡尔积
- 物化常用中间结果
"""

# 3. Spark 优化
spark_optimization_tips = """
- 调整 executor 内存和核心数
- 使用 broadcast join 处理小表
- 缓存重复使用的 DataFrame
- 避免数据倾斜 (salting 技术)
- 使用 Tungsten 引擎
"""

print("数据工程性能优化技巧:")
print("=" * 60)
print(batch_optimization_tips)
print(sql_optimization_tips)
print(spark_optimization_tips)

---

🎯 学习路径

入门阶段 (1-3 个月)

第 1 月：编程与 SQL 基础
  - Python 高级编程
  - SQL 进阶 (窗口函数、CTE)
  - Linux 基础命令
  - Git 版本控制

第 2 月：数据库与 ETL 基础
  - 关系数据库原理
  - ETL 设计模式
  - Apache Airflow 基础
  - 数据仓库概念

第 3 月：大数据基础
  - Hadoop 生态系统
  - Apache Spark 基础
  - 分布式计算原理
  - 第一个数据管道项目

进阶阶段 (3-6 个月)

第 4-5 月：流处理与云原生
  - Apache Kafka
  - Apache Flink
  - 云数据平台 (AWS/GCP/Azure)
  - dbt 数据转换

第 6 月：数据治理与架构
  - 数据建模 (维度建模)
  - 数据质量管理
  - 数据治理框架
  - 数据湖架构

---

📚 中文资源

博客与社区

资源	类型	链接
数据工程实战	公众号	链接
大数据工匠	博客	链接
DataFun	社区	链接

视频课程

课程	讲师	平台	链接
大数据入门	尚硅谷	B 站	链接
Spark 进阶	李兴华	慕课网	链接

---

最后更新: 2026-06-01

相关文档: - big-data-business-resources.md - 大数据与商业分析 - sql-resources.md - SQL 与数据库分析 - machine-learning-resources.md - 机器学习资源

---

**数据工程资源专题 | 构建可靠的数据基础设施** [返回顶部](#数据工程资源专题)