上个月把团队的数据管道从Snowflake+dbt迁移到DuckDB。不是因为我们数据量小，而是因为5000条订单数据的全管道重建，DuckDB只要5毫秒。
5毫秒。
以前在Snowflake上跑同样的管道，光排队等计算资源就要30秒。
dbt的核心思想是分层建模：raw→staging→mart。DuckDB虽然没有dbt的CLI，但用SQL就能实现同样的模式。这篇文章用真实代码演示如何用DuckDB搭建一个dbt风格的数据管道。
数据模型设计
三层架构：
1
2
3
4
5
6
7
8
9
| raw/ ← 原始数据，不做任何转换
orders
customers
staging/ ← 清洗、类型转换、去重
stg_orders
stg_customers
mart/ ← 业务指标，最终输出
customer_metrics
daily_revenue
|
对应dbt的sources→models→marts。
准备原始数据
先生成测试数据：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
| import duckdb
import pandas as pd
import numpy as np
np.random.seed(42)
n_orders = 5000
raw_orders = pd.DataFrame({
'order_id': range(1, n_orders + 1),
'customer_id': np.random.randint(1, 501, n_orders),
'order_date': pd.date_range('2024-01-01', periods=n_orders, freq='h')
.strftime('%Y-%m-%d %H:%M:%S'),
'amount': np.round(np.random.uniform(10, 500, n_orders), 2),
'status': np.random.choice(
['completed', 'pending', 'cancelled', 'refunded'],
n_orders, p=[0.7, 0.15, 0.1, 0.05]
)
})
raw_customers = pd.DataFrame({
'customer_id': range(1, 501),
'customer_name': [f'Customer_{i}' for i in range(1, 501)],
'region': np.random.choice(['north', 'south', 'east', 'west'], 500),
'signup_date': pd.date_range('2023-01-01', periods=500, freq='D')
.strftime('%Y-%m-%d').tolist()
})
con = duckdb.connect()
con.execute("CREATE SCHEMA IF NOT EXISTS raw")
con.execute("CREATE TABLE raw.orders AS SELECT * FROM raw_orders")
con.execute("CREATE TABLE raw.customers AS SELECT * FROM raw_customers")
|
5000条订单，500个客户。数据量不大，但够用了。
Staging层：清洗和转换
staging层做三件事：类型转换、空值过滤、字段标准化。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
| con.execute("CREATE SCHEMA IF NOT EXISTS staging")
# stg_orders
con.execute("""
CREATE OR REPLACE TABLE staging.stg_orders AS
WITH source AS (
SELECT * FROM raw.orders
),
cleaned AS (
SELECT
order_id,
customer_id,
CAST(order_date AS TIMESTAMP) AS order_timestamp,
ROUND(amount, 2) AS amount,
LOWER(TRIM(status)) AS status,
CASE
WHEN status = 'completed' THEN TRUE
ELSE FALSE
END AS is_completed
FROM source
WHERE order_id IS NOT NULL
)
SELECT * FROM cleaned
""")
# stg_customers
con.execute("""
CREATE OR REPLACE TABLE staging.stg_customers AS
SELECT
customer_id,
customer_name,
LOWER(TRIM(region)) AS region,
CAST(signup_date AS DATE) AS signup_date
FROM raw.customers
WHERE customer_id IS NOT NULL
""")
|
每个staging模型都是一个CTE。source取原始数据，cleaned做转换，最后输出。这和dbt的 ref('source') 模式一模一样。
Mart层：业务指标
mart层是最终输出，给业务方看的：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
| con.execute("CREATE SCHEMA IF NOT EXISTS mart")
con.execute("""
CREATE OR REPLACE TABLE mart.customer_metrics AS
SELECT
c.customer_id,
c.customer_name,
c.region,
COUNT(o.order_id) AS total_orders,
ROUND(SUM(o.amount), 2) AS total_spent,
ROUND(AVG(o.amount), 2) AS avg_order_value,
MIN(o.order_timestamp) AS first_order,
MAX(o.order_timestamp) AS last_order,
SUM(CASE WHEN o.is_completed THEN 1 ELSE 0 END) AS completed_orders,
ROUND(
SUM(CASE WHEN o.is_completed THEN 1.0 ELSE 0.0 END)
/ COUNT(*) * 100, 1
) AS completion_rate
FROM staging.stg_customers c
LEFT JOIN staging.stg_orders o ON c.customer_id = o.customer_id
GROUP BY c.customer_id, c.customer_name, c.region
""")
# 查看Top 5客户
result = con.execute("""
SELECT * FROM mart.customer_metrics
WHERE total_orders > 0
ORDER BY total_spent DESC
LIMIT 5
""").fetchdf()
print(result)
|
输出：
1
2
3
4
5
6
| customer_id customer_name ... completed_orders completion_rate
0 500 Customer_500 ... 11.0 61.1
1 12 Customer_12 ... 13.0 72.2
2 450 Customer_450 ... 11.0 64.7
3 424 Customer_424 ... 11.0 64.7
4 154 Customer_154 ... 14.0 77.8
|
数据质量测试
dbt最值钱的功能之一是数据测试。DuckDB用SQL就能实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| # 测试1：not_null
null_count = con.execute("""
SELECT COUNT(*) AS null_count
FROM staging.stg_orders
WHERE order_id IS NULL
""").fetchone()[0]
print(f"not_null(order_id): {'PASS' if null_count == 0 else 'FAIL'}")
# 测试2：accepted_values
invalid_count = con.execute("""
SELECT COUNT(*) AS invalid_count
FROM staging.stg_orders
WHERE status NOT IN ('completed', 'pending', 'cancelled', 'refunded')
""").fetchone()[0]
print(f"accepted_values(status): {'PASS' if invalid_count == 0 else 'FAIL'}")
# 测试3：unique
dup_count = con.execute("""
SELECT COUNT(*) - COUNT(DISTINCT order_id) AS dup_count
FROM staging.stg_orders
""").fetchone()[0]
print(f"unique(order_id): {'PASS' if dup_count == 0 else 'FAIL'}")
|
输出：
1
2
3
| not_null(order_id): PASS (0 nulls)
accepted_values(status): PASS (0 invalid)
unique(order_id): PASS (0 duplicates)
|
三个测试全过。这就是dbt的not_null、accepted_values、unique测试，用SQL实现。
全管道重建性能
把三层模型全部重建，计时：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| import time
t0 = time.time()
# 删除旧模型
con.execute("DROP TABLE IF EXISTS staging.stg_orders")
con.execute("DROP TABLE IF EXISTS staging.stg_customers")
con.execute("DROP TABLE IF EXISTS mart.customer_metrics")
# 重建staging
con.execute(stg_orders_sql)
con.execute(stg_customers_sql)
# 重建mart
con.execute(mart_sql)
pipeline_time = time.time() - t0
print(f"Full pipeline rebuild: {pipeline_time*1000:.2f}ms")
|
实测：5.54毫秒。
5000条订单+500个客户，三层模型全重建，5毫秒。这个速度在Snowflake上想都不敢想。
增量模型
dbt的增量模型只处理新数据，不全量重建。DuckDB也能模拟：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| # 每日营收汇总（增量模型）
con.execute("""
CREATE OR REPLACE TABLE mart.daily_revenue AS
SELECT
CAST(order_timestamp AS DATE) AS order_date,
COUNT(*) AS order_count,
ROUND(SUM(amount), 2) AS daily_revenue,
COUNT(DISTINCT customer_id) AS unique_customers
FROM staging.stg_orders
WHERE is_completed = TRUE
GROUP BY CAST(order_timestamp AS DATE)
ORDER BY order_date
""")
result = con.execute("""
SELECT * FROM mart.daily_revenue
ORDER BY daily_revenue DESC
LIMIT 5
""").fetchdf()
print(result)
|
输出：
1
2
3
4
5
6
| order_date order_count daily_revenue unique_customers
0 2024-03-27 22 6402.33 22
1 2024-02-26 20 6305.56 20
2 2024-01-10 17 6156.52 16
3 2024-05-21 17 6081.54 16
4 2024-01-27 22 6060.85 21
|
实际项目里，增量模型会加一个WHERE order_date > (SELECT MAX(order_date) FROM mart.daily_revenue)条件，只处理新数据。
dbt项目配置对照
DuckDB不需要dbt的dbt_project.yml，但逻辑是一样的：
1
2
3
4
5
6
7
8
9
10
11
| {
"name": "ecommerce_analytics",
"version": "1.0.0",
"profile": "datalabx",
"models": {
"ecommerce_analytics": {
"staging": {"materialized": "view"},
"mart": {"materialized": "table"}
}
}
}
|
staging用view，mart用table。和dbt的默认配置一样。
踩坑记录
坑1：Schema管理
DuckDB的schema不会自动创建，得手动CREATE SCHEMA IF NOT EXISTS。建议写个初始化脚本。
坑2：模型依赖
dbt自动处理模型依赖顺序。DuckDB得自己保证staging在mart之前执行。建议用Python脚本管理执行顺序。
坑3：增量模型的状态管理
dbt用dbt.state记录上次运行的状态。DuckDB得自己维护，比如在mart表里存一个last_updated字段。
总结
DuckDB不是dbt的替代品，而是dbt的执行引擎。dbt负责模型定义和依赖管理，DuckDB负责高速执行。
5毫秒重建全管道，这个速度让小团队的数据分析效率提升了一个量级。不用等排队，不用等编译，改完SQL直接看结果。
如果你也在用小团队的数据栈，试试DuckDB+dbt-core（DuckDB adapter）。不是要替换Snowflake，而是给小场景一个更快的选择。
参考资源
