AI 时代的数据工程实战：dbt + Airflow + DuckDB，3 步搭建现代数据栈

数据工程师的日子越来越不好做了——业务要实时报表、AI 团队要训练数据集、老板要 BI 看板。传统 ETL 跑不动了？这篇文章带你用 dbt + Airflow + DuckDB 搭一套轻量级但够用的现代数据栈，Python 全家桶，代码可运行。

---

一、传统数据工程 vs 现代数据栈

先说清楚我们为什么需要这套组合：

维度	传统 ETL（Kettle/SSIS）	现代数据栈
开发方式	拖拽界面	代码（SQL/Python）
版本控制	❌ 几乎不可能	✅ Git 原生支持
测试	手动跑一遍看结果	✅ 自动化单元测试
文档	写 Word/Confluence	✅ 自动生成 Data Catalog
部署	人工操作	✅ CI/CD 流水线
成本	商业授权贵	开源免费为主
学习曲线	低，但要靠”经验”	中等，但有标准范式

核心变化：从”ETL”到”ELT”——先加载原始数据，再在数据仓库里做转换。这样能充分利用数据库/数仓的计算能力，而且转换逻辑全部代码化。

---

二、架构总览

┌─────────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│  数据源      │ →  │  Airflow │ →  │  DuckDB  │ →  │   dbt    │
│ (API/文件)  │    │ (调度)    │    │ (存储)    │    │ (转换)    │
└─────────────┘    └──────────┘    └──────────┘    └──────────┘
                                                             ↓
                                                      ┌──────────┐
                                                      │ BI/AI 输出│
                                                      │ (报表/特征)│
                                                      └──────────┘

• Airflow：任务调度引擎，编排数据采集和模型训练流程
• DuckDB：嵌入式 OLAP 数据库，单文件部署，查询速度飞快
• dbt（Data Build Tool）：声明式 SQL 转换框架，自动管理依赖和文档

---

三、Step 1 — DuckDB 快速上手

DuckDB 是近年最火的开源分析型数据库之一。它不需要安装服务端，直接嵌入 Python 进程：

# duckdb_setup.py
import duckdb
import pandas as pd

# 创建/连接数据库（就是一个文件）
conn = duckdb.connect("data_warehouse.db")

# 建表并导入示例电商订单数据
conn.execute("""
    CREATE TABLE IF NOT EXISTS raw_orders (
        order_id VARCHAR PRIMARY KEY,
        customer_id VARCHAR,
        product_name VARCHAR,
        amount DECIMAL(10,2),
        category VARCHAR,
        order_date DATE,
        status VARCHAR
    )
""")

# 用 Python 插入一批测试数据
sample_orders = [
    ("ORD001", "C001", "腾讯云服务器年付", 1200.00, "云服务", "2026-05-01", "completed"),
    ("ORD002", "C002", "阿里云OSS存储包", 299.00, "存储", "2026-05-02", "completed"),
    ("ORD003", "C001", "大模型API调用费", 580.50, "AI服务", "2026-05-03", "processing"),
    ("ORD004", "C003", "华为云CDN加速", 800.00, "网络", "2026-05-04", "completed"),
    ("ORD005", "C002", "域名续费一年", 68.00, "基础服务", "2026-05-05", "cancelled"),
    ("ORD006", "C004", "SSL证书采购", 199.00, "安全", "2026-05-06", "completed"),
    ("ORD007", "C001", "对象存储流量包", 150.00, "存储", "2026-05-07", "completed"),
    ("ORD008", "C003", "企业邮箱套餐", 3600.00, "办公", "2026-05-08", "completed"),
    ("ORD009", "C005", "负载均衡实例", 450.00, "网络", "2026-05-09", "processing"),
    ("ORD010", "C004", "数据库备份服务", 200.00, "数据库", "2026-05-10", "completed"),
]

conn.executemany(
    "INSERT INTO raw_orders VALUES (?, ?, ?, ?, ?, ?, ?)",
    sample_orders
)

print(f"已插入 {len(sample_orders)} 条订单数据")

# DuckDB 的杀手锏——超快聚合查询
result = conn.sql("""
    SELECT 
        category AS 产品分类,
        COUNT(*) AS 订单数,
        ROUND(SUM(amount), 2) AS 总金额,
        ROUND(AVG(amount), 2) AS 平均客单价
    FROM raw_orders
    WHERE status = 'completed'
    GROUP BY category
    ORDER BY 总金额 DESC
""").df()

print("\n=== 销售额按分类统计 ===")
print(result)

# 导出给 AI 训练用
result.to_csv("features/sales_by_category.csv", index=False)
print("\n特征文件已导出至 features/sales_by_category.csv")

为什么选 DuckDB 而不是 PostgreSQL？

场景	DuckDB	PostgreSQL
单机数据分析（<100GB）	⭐⭐⭐ 极快	⚠️ 还行
多用户并发写入	⚠️ 不擅长	⭐⭐⭐ 强项
嵌入式/无服务端部署	✅ 一个 .db 文件	❌ 需要进程
Python/Pandas 集成	⭐⭐⭐ 原生	⚠️ 需 psycopg2
列式存储查询性能	⭐⭐⭐ 比内存还快	⚠️ 行存为主

对于中小企业的数据仓库场景（日增 < 100GB），DuckDB 完全够用，而且零运维成本。

---

四、Step 2 — dbt 数据转换

4.1 项目初始化

pip install dbt-core dbt-duckdb
mkdir my_data_project && cd my_data_project
dbt init my_data_project --adapter duckdb
cd my_data_project

4.2 配置 profiles.yml

# ~/.dbt/profiles.yml
my_data_project:
  outputs:
    dev:
      type: duckdb
      path: ../data_warehouse.db
      threads: 4
  target: dev

4.3 编写 dbt Model（核心！）

-- models/stg_orders.sql
-- 清洗层（Staging）：对原始数据做标准化处理

SELECT
    order_id,
    customer_id,
    TRIM(product_name) AS product_name,
    amount,
    CASE
        WHEN category IN ('云服务', '网络', '数据库') THEN '基础设施'
        WHEN category = 'AI服务' THEN 'AI与智能'
        WHEN category = '存储' THEN '存储与备份'
        ELSE '其他'
    END AS category_group,
    order_date,
    status
FROM {{ source('raw', 'raw_orders') }}
WHERE status != 'cancelled'

-- models/fct_daily_revenue.sql
-- 聚合层（Fact）：按天统计营收

WITH stg_orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
)

SELECT
    order_date AS date_key,
    COUNT(DISTINCT order_id) AS total_orders,
    COUNT(DISTINCT customer_id) AS active_customers,
    ROUND(SUM(amount), 2) AS daily_gmv,
    ROUND(AVG(amount), 2) AS avg_order_value
FROM stg_orders
GROUP BY order_date
ORDER BY date_key

-- models/fct_customer_rfm.sql
-- RFM 分析：客户价值分层

WITH customer_metrics AS (
    SELECT
        customer_id,
        MIN(order_date) AS first_order_date,
        MAX(order_date) AS last_order_date,
        COUNT(*) AS frequency,
        SUM(amount) AS monetary
    FROM {{ ref('stg_orders') }}
    GROUP BY customer_id
)

SELECT
    *,
    CASE
        WHEN monetary >= 3000 AND frequency >= 2 THEN 'VIP客户'
        WHEN monetary >= 500 THEN '高价值客户'
        WHEN frequency >= 1 THEN '普通客户'
        ELSE '新客户/流失风险'
    END AS customer_segment
FROM customer_metrics
ORDER BY monetary DESC

4.4 定义 Source 和 Tests

# models/schema.yml

version: 2

sources:
  - name: raw
    tables:
      - name: raw_orders
        description: 原始订单表
        columns:
          - name: order_id
            description: 订单ID
            tests:
              - unique
              - not_null

models:
  - name: stg_orders
    description: 订单清洗层
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
          - relationships:
              to: ref('stg_orders')
              field: order_id

  - name: fct_daily_revenue
    description: 日营收事实表
    tests:
      - dbt_utils.expression_is_true:
          expression: "daily_gmv >= 0"

  - name: fct_customer_rfm
    description: 客户RFM价值分层表

4.5 运行 dbt

# 运行所有 model 并生成文档
dbt run
dbt test       # 执行自动化测试
dbt docs generate && dbt docs serve  # 启动本地文档站

---

五、Step 3 — Airflow 任务调度

5.1 安装与配置

pip install apache-airflow
export AIRFLOW_HOME=~/airflow_home
airflow db init  # 初始化元数据库

5.2 DAG 定义

# dags/data_pipeline_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.common.sql.operators.sql import SqlOperator
import duckdb
import subprocess
import logging

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    dag_id='daily_data_pipeline',
    default_args=default_args,
    start_date=datetime(2026, 5, 1),
    schedule_interval='0 8 * * *',     # 每天 8 点执行
    catchup=False,
    tags=['data-engineering', 'production'],
)


def extract_task(**context):
    """步骤1：数据抽取——模拟从业务系统拉取数据"""
    logging.info("📥 开始数据抽取...")

    # 这里可以换成真实的 API 调用或数据库同步
    # 示例：从 CSV/API 加载数据到 DuckDB
    conn = duckdb.connect("data_warehouse.db")

    # 模拟增量抽取：只取昨天的数据
    yesterday = context['execution_date'].strftime('%Y-%m-%d')
    logging.info(f"抽取日期范围: {yesterday}")

    # 实际项目中这里对接你的业务数据库/API
    # conn.execute("INSERT INTO raw_orders SELECT * FROM staging WHERE order_date = ?", [yesterday])

    conn.close()
    return f"✅ 抽取完成，目标日期: {yesterday}"


def transform_task():
    """步骤2：执行 dbt 转换"""
    logging.info("🔄 开始 dbt 转换...")
    result = subprocess.run(
        ["dbt", "run", "--project-dir", "/path/to/my_data_project"],
        capture_output=True,
        text=True,
        cwd="/path/to/my_data_project"
    )
    if result.returncode != 0:
        raise Exception(f"dbt run 失败: {result.stderr}")
    logging.info("✅ dbt 转换完成")


def load_to_ai_features():
    """步骤3：导出 AI 特征数据"""
    logging.info("🤖 开始生成 AI 特征数据...")

    conn = duckdb.connect("data_warehouse.db")

    # 导出 RFM 分层结果供推荐模型使用
    rfm_data = conn.sql("""
        SELECT * FROM fct_customer_rfm
        ORDER BY monetary DESC
    """).df()

    rfm_data.to_csv("/mnt/ml_features/customer_rfm.csv", index=False)
    logging.info(f"✅ 已导出 {len(rfm_data)} 条客户特征")

    # 导出每日趋势供预测模型使用
    trend_data = conn.sql("""
        SELECT * FROM fct_daily_revenue
    """).df()

    trend_data.to_csv("/mnt/ml_features/daily_trend.csv", index=False)
    logging.info(f"✅ 已导出 {len(trend_data)} 条趋势特征")

    conn.close()


def notify_success(context):
    """通知任务成功"""
    logging.info("🎉 今日数据流水线执行完毕!")


# 定义 Task 及依赖关系
with dag:
    t1 = PythonOperator(
        task_id="extract_data",
        python_callable=extract_task,
    )
    t2 = PythonOperator(
        task_id="run_dbt_transform",
        python_callable=transform_task,
    )
    t3 = PythonOperator(
        task_id="export_ai_features",
        python_callable=load_to_ai_features,
    )
    t4 = PythonOperator(
        task_id="notify",
        python_callable=notify_success,
    )

    # 依赖链：抽取 → 转换 → 特征导出 → 通知
    t1 >> t2 >> t3 >> t4

5.3 启动 Airflow

# 启动 Web UI（默认 http://localhost:8080）
 airflow webserver --port 8080 &
 airflow scheduler

---

六、完整对比：这套方案 vs 传统方案

维度	Kettle + MySQL	dbt + Airflow + DuckDB	Spark + Hive
搭建时间	半天	1 小时	3 天+
运维成本	低	极低	高（集群维护）
数据量上限	百万行	亿行（单机）	PB 级
团队协作	❌	✅ Git + PR Review	✅
文档自动生成	❌	✅ dbt docs	⚠️ 部分
适合团队规模	1~3 人	3~20 人	20+ 人

结论：如果你的团队在 20 人以内、日数据量在亿级行以下，dbt + Airflow + DuckDB 这套方案是性价比最高的选择。不需要 Hadoop，不需要 Spark 集群，一台云服务器就能跑起来。

---

总结

现代数据工程的精髓不在于工具多牛，而在于可复现、可测试、可追溯。

• DuckDB 替代你那台跑得半死不活的 MySQL 分析库
• dbt 把散落各处的 SQL 变成有版本管理的代码资产
• Airflow 让每天凌晨 3 点手动跑脚本的日子成为历史

这三样东西加在一起，就是中小团队最实用的数据基建方案。别再犹豫了，今天就开始搭吧！

---

👤 作者简介

一枚在大中原腹地（河南）卖公有云的从业者，主营腾讯云/阿里云/华为云，曾踩坑无数，现专注AI大模型应用落地。关注公众号「AI热点日报」，围观AI前沿动态~

博客：y1cloud.cn

---