第18课：dbt数据转换

📦 dbt数据转换

dbt是ELT范式的核心工具，让数据分析师用SQL定义转换逻辑在数仓中执行。

🐍 Python实战：核心实战

# dbt模拟引擎
import sqlite3, time, random, json
from datetime import datetime
from collections import defaultdict

conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
c = conn.cursor()

# 创建源表
c.execute('CREATE TABLE raw_orders (id INTEGER PRIMARY KEY, customer_id TEXT, amount REAL, status TEXT, region TEXT, order_date TEXT)')
for i in range(1000):
    c.execute("INSERT INTO raw_orders VALUES (?,?,?,?,?,?)",
        (i+1, f"CUST_{random.randint(1,200)}", round(random.uniform(10,5000),2),
         random.choice(['pending','shipped','delivered','cancelled']),
         random.choice(['华东','华南','华北']), datetime.now().strftime('%Y-%m-%d')))
conn.commit()

class dbtModel:
    def __init__(self, name, sql, materialized='table', depends_on=None):
        self.name = name; self.sql = sql; self.materialized = materialized
        self.depends_on = depends_on or []
    def execute(self, conn):
        start = time.perf_counter()
        conn.execute(self.sql)
        conn.commit()
        ms = (time.perf_counter()-start)*1000
        return {'name':self.name,'materialized':self.materialized,'ms':ms}

class dbtProject:
    def __init__(self, name):
        self.name = name; self.models = []
    def add_model(self, model): self.models.append(model)
    def run(self):
        print(f"dbt run: {self.name}")
        executed = set()
        for model in self.models:
            if all(d in executed for d in model.depends_on):
                result = model.execute(conn)
                executed.add(model.name)
                print(f"  ✅ {result['name']} ({result['materialized']}): {result['ms']:.1f}ms")
    def test(self):
        print(f"dbt test: {self.name}")
        tests = [
            ("unique_raw_orders_id", "SELECT COUNT(*)-COUNT(DISTINCT id) FROM raw_orders"),
            ("not_null_raw_orders_amount", "SELECT COUNT(*) FROM raw_orders WHERE amount IS NULL"),
            ("accepted_values_status", "SELECT COUNT(*) FROM raw_orders WHERE status NOT IN ('pending','shipped','delivered','cancelled')"),
        ]
        for name, sql in tests:
            result = conn.execute(sql).fetchone()[0]
            status = "✅ PASS" if result == 0 else f"❌ FAIL ({result})"
            print(f"  {name}: {status}")

# 构建dbt项目
project = dbtProject("my_project")
project.add_model(dbtModel("stg_orders",
    "CREATE TABLE stg_orders AS SELECT id, customer_id, amount, status, region, order_date FROM raw_orders WHERE status != 'cancelled'",
    "table", []))
project.add_model(dbtModel("int_customer_orders",
    "CREATE TABLE int_customer_orders AS SELECT customer_id, COUNT(*) as order_count, SUM(amount) as total_amount FROM stg_orders GROUP BY customer_id",
    "table", ["stg_orders"]))
project.add_model(dbtModel("fct_daily_sales",
    "CREATE TABLE fct_daily_sales AS SELECT order_date, region, COUNT(*) as orders, SUM(amount) as revenue FROM stg_orders GROUP BY order_date, region",
    "table", ["stg_orders"]))

project.run()
project.test()

# 查询结果
print("\n客户订单汇总(前5):")
for row in c.execute("SELECT * FROM int_customer_orders ORDER BY total_amount DESC LIMIT 5"):
    r = dict(row); print(f"  {r['customer_id']}: {r['order_count']}单, ¥{r['total_amount']:,.0f}")
print("✅ 验证通过 - dbt模拟引擎运行正常")

📖 原理深入：数据编排与治理体系

🏆 最佳实践：数据编排与治理检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答