transforming-data

数据转换

使用现代转换模式、框架和编排工具，将原始数据转化为分析资产。

目的

在现代数据技术栈中选择并实现数据转换模式。使用 SQL（dbt）、Python DataFrame（pandas、polars、PySpark）和流水线编排（Airflow、Dagster、Prefect），将原始数据转化为干净、经过测试且有文档说明的分析数据集。

适用场景

在以下情况下使用此技能：

• 在 ETL 和 ELT 转换模式之间进行选择
• 构建 dbt 模型（staging、intermediate、marts）
• 实现增量数据加载和合并策略
• 将 pandas 代码迁移到 polars，以提升性能
• 编排具有依赖关系和重试机制的数据流水线
• 添加数据质量测试和验证
• 使用 PySpark 处理大型数据集
• 创建可用于生产的转换工作流

快速入门：常见模式

dbt 增量模型

{{
  config(
    materialized='incremental',
    unique_key='order_id'
  )
}}

select order_id, customer_id, order_created_at, sum(revenue) as total_revenue
from {{ ref('int_order_items_joined') }}
group by 1, 2, 3

{% if is_incremental() %}
    where order_created_at > (select max(order_created_at) from {{ this }})
{% endif %}

polars 高性能转换

import polars as pl

result = (
    pl.scan_csv('large_dataset.csv')
    .filter(pl.col('year') == 2024)
    .with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
    .group_by('region')
    .agg(pl.col('revenue').sum())
    .collect()  # Execute lazy query
)

Airflow 数据流水线

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

with DAG(
    dag_id='daily_sales_pipeline',
    schedule_interval='0 2 * * *',
    default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)},
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    extract = PythonOperator(task_id='extract', python_callable=extract_data)
    transform = PythonOperator(task_id='transform', python_callable=transform_data)
    extract >> transform

决策框架

ETL 与 ELT 的选择

在以下情况下使用 ELT（提取、加载、转换）：

• 使用现代云数据仓库（Snowflake、BigQuery、Databricks）
• 转换逻辑频繁变化
• 团队中有 SQL 分析师
• 数据量为 10GB-1TB 以上（利用数据仓库的并行处理能力）

工具：dbt、Dataform、Snowflake tasks、BigQuery scheduled queries

在以下情况下使用 ETL（提取、转换、加载）：

• 法规合规要求在加载前对数据进行脱敏（PII/PHI）
• 目标系统缺少计算能力
• 需要实时流处理并立即转换
• 使用不具备云数据仓库的旧系统

工具：AWS Glue、Azure Data Factory、自定义 Python 脚本

在敏感数据清理（ETL）与分析转换（ELT）结合使用时，采用混合模式。

默认建议：使用 ELT 和 dbt，除非特定的合规或性能限制要求采用 ETL。

如需了解详细模式，请参阅 references/etl-vs-elt-patterns.md。

DataFrame 库的选择

在以下情况下选择 pandas：

• 数据量小于 500MB
• 进行原型开发或探索性分析
• 需要兼容仅支持 pandas 的库

在以下情况下选择 polars：

• 数据量为 500MB-100GB
• 对性能要求较高（比 pandas 快 10-100 倍）
• 用于受内存限制的生产流水线
• 希望借助查询优化进行惰性求值

在以下情况下选择 PySpark：

• 数据量大于 100GB
• 需要在集群中进行分布式处理
• 已有 Spark 基础设施（EMR、Databricks）

迁移路径：pandas → polars（更简单，API 相似）或 pandas → PySpark（需要集群）

如需查看对比和迁移指南，请参阅 references/dataframe-comparison.md。

编排工具的选择

在以下情况下选择 Airflow：

• 企业级生产环境（已在大规模场景中得到验证）
• 需要 5,000 多种集成
• 提供托管服务（AWS MWAA、GCP Cloud Composer）

在以下情况下选择 Dagster：

• 大量使用 dbt（原生 dbt_assets 集成）
• 优先考虑数据沿袭和基于资产的工作流
• 需要可测试性的 ML 流水线

在以下情况下选择 Prefect：

• 动态工作流（运行时生成任务）
• 倾向于云原生架构
• 使用带装饰器的 Pythonic API

稳妥的默认选择：Airflow（经实践验证），除非有特定需求需要使用 Dagster/Prefect。

如需了解详细模式，请参阅 references/orchestration-patterns.md。

使用 dbt 进行 SQL 转换

模型分层结构

1. 暂存层（models/staging/）

   • 与源表保持 1:1 对应
   • 仅进行少量转换（重命名、类型转换、基本筛选）
   • 物化为视图或临时模型

2. 中间层（models/intermediate/）

   • 包含业务逻辑和复杂联接
   • 不向最终用户开放
   • 通常为临时模型（仅 CTE）

3. 集市层（models/marts/）

   • 用于报告的最终模型
   • 事实表（事件、交易）
   • 维度表（客户、产品）
   • 物化为表或增量模型

dbt 物化类型

视图：每次引用模型时都会重新运行查询。适用于快速查询和暂存层。

表：每次运行都会完全刷新。适用于经常查询的模型和计算开销较大的场景。

增量：仅处理新增或变更的记录。适用于大型事实表和事件日志。

临时模型：仅使用 CTE，不会持久化。适用于中间计算。

dbt 测试

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: total_revenue
        tests:
          - dbt_utils.accepted_range:
              min_value: 0

如需了解全面的 dbt 模式，请参阅：

• references/dbt-best-practices.md
• references/incremental-strategies.md

Python DataFrame 转换

pandas 转换

import pandas as pd

df = pd.read_csv('sales.csv')
result = (
    df
    .query('year == 2024')
    .assign(revenue=lambda x: x['quantity'] * x['price'])
    .groupby('region')
    .agg({'revenue': ['sum', 'mean']})
)

polars 转换（快 10-100 倍）

import polars as pl

result = (
    pl.scan_csv('sales.csv')  # Lazy evaluation
    .filter(pl.col('year') == 2024)
    .with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
    .group_by('region')
    .agg([
        pl.col('revenue').sum().alias('revenue_sum'),
        pl.col('revenue').mean().alias('revenue_mean')
    ])
    .collect()  # Execute lazy query
)

主要区别：

• polars 使用 scan_csv()（惰性），而 pandas 使用 read_csv()（急切）
• polars 使用 with_columns()，而 pandas 使用 assign()
• polars 使用 pl.col() 表达式，而 pandas 使用字符串引用
• polars 需要调用 collect() 才会执行惰性查询

用于分布式处理的 PySpark

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("Transform").getOrCreate()
df = spark.read.csv('sales.csv', header=True, inferSchema=True)

result = (
    df
    .filter(F.col('year') == 2024)
    .withColumn('revenue', F.col('quantity') * F.col('price'))
    .groupBy('region')
    .agg(F.sum('revenue').alias('total_revenue'))
)

如需迁移指南，请参阅 references/dataframe-comparison.md。

流水线编排

Airflow DAG 结构

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-engineering',
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    dag_id='data_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # Daily at 2 AM
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    task1 = PythonOperator(task_id='extract', python_callable=extract_fn)
    task2 = PythonOperator(task_id='transform', python_callable=transform_fn)
    task1 >> task2  # Define dependency

任务依赖模式

线性：A >> B >> C（顺序执行） 扇出：A >> [B, C, D]（A 之后并行执行） 扇入：[A, B, C] >> D（D 等待所有任务完成）

如需了解 Airflow、Dagster 和 Prefect 的模式，请参阅 references/orchestration-patterns.md。

数据质量与测试

dbt 测试

通用测试（可复用）：unique、not_null、accepted_values、relationships

单一测试（自定义 SQL）：

-- tests/assert_positive_revenue.sql
select * from {{ ref('fct_orders') }}
where total_revenue < 0

Great Expectations

import great_expectations as gx

context = gx.get_context()
suite = context.add_expectation_suite("orders_suite")

suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="total_revenue", min_value=0
    )
)

如需了解全面的测试模式，请参阅 references/data-quality-testing.md。

高级 SQL 模式

用于分析的窗口函数：

select
    order_date,
    daily_revenue,
    avg(daily_revenue) over (
        partition by region
        order by order_date
        rows between 6 preceding and current row
    ) as revenue_7d_ma,
    sum(daily_revenue) over (
        partition by region
        order by order_date
    ) as cumulative_revenue
from daily_sales

如需了解高级窗口函数，请参阅 references/window-functions-guide.md。

生产环境最佳实践

幂等性

确保多次运行转换后产生相同结果：

• 在增量模型中使用 merge 语句
• 实现去重逻辑
• 在 dbt 增量模型中使用 unique_key

增量加载

{% if is_incremental() %}
    where created_at > (select max(created_at) from {{ this }})
{% endif %}

错误处理

try:
    result = perform_transformation()
    validate_result(result)
except ValidationError as e:
    log_error(e)
    raise

监控

• 设置 Airflow 任务失败时的邮件/Slack 告警
• 监控 dbt 测试失败情况
• 跟踪数据新鲜度（SLA）
• 记录行数和数据质量指标

工具推荐

SQL 转换：dbt Core（行业标准，支持多种数据仓库，生态丰富）

pip install dbt-core dbt-snowflake

Python DataFrame：polars（比 pandas 快 10-100 倍，支持多线程和惰性求值）

pip install polars

编排：Apache Airflow（经过大规模实战验证，拥有 5,000+ 项集成）

pip install apache-airflow

示例

可运行示例位于：

• examples/python/pandas-basics.py - pandas 转换
• examples/python/polars-migration.py - 从 pandas 迁移到 polars
• examples/python/pyspark-transformations.py - PySpark 操作
• examples/python/airflow-data-pipeline.py - 完整的 Airflow DAG
• examples/sql/dbt-staging-model.sql - dbt 暂存层
• examples/sql/dbt-intermediate-model.sql - dbt 中间层
• examples/sql/dbt-incremental-model.sql - 增量模式
• examples/sql/window-functions.sql - 高级 SQL

脚本

• scripts/generate_dbt_models.py - 生成 dbt 模型样板代码
• scripts/benchmark_dataframes.py - 比较 pandas 与 polars 的性能

相关技能

如需了解数据摄取模式，请参阅 ingesting-data。 如需了解数据可视化，请参阅 visualizing-data。 如需了解数据库设计，请参阅 databases-* 技能。 如需了解实时流处理，请参阅 streaming-data。 如需了解数据平台架构，请参阅 ai-data-engineering。 如需了解流水线监控，请参阅 observability。