DuckDB 宏（Macros）实战：把重复 SQL 封装成函数，效率翻 10 倍

import duckdb

# 创建示例数据
conn = duckdb.connect()
conn.execute("""
    CREATE TABLE orders AS
    SELECT * FROM (VALUES
        (1, 'A', 100.0, 0, '2023-10-01'),
        (2, 'B', 200.0, 1, '2023-10-01'),  -- 测试单
        (3, 'A', 300.0, 0, '2023-10-02'),
        (4, 'C', 400.0, 1, '2023-10-02'),  -- 退款单
        (5, 'B', 500.0, 0, '2023-10-03')
    ) AS t(order_id, seller, amount, is_refund, order_date)
""")

# 定义宏：计算有效 GMV
conn.execute("""
    CREATE OR REPLACE MACRO effective_gmv(amount, is_refund) AS
        CASE 
            WHEN is_refund = 0 THEN amount 
            ELSE 0 
        END;
""")

# 使用宏进行每日汇总
result = conn.execute("""
    SELECT 
        order_date,
        SUM(effective_gmv(amount, is_refund)) AS daily_gmv
    FROM orders
    GROUP BY order_date
    ORDER BY order_date
""").fetchdf()

print(result)

import duckdb

conn = duckdb.connect()

# 创建两张模拟的日分区表
for date in ['20231001', '20231002']:
    conn.execute(f"""
        CREATE TABLE sales_{date} AS
        SELECT * FROM (VALUES
            ('Product_A', 100 + {date[-2:]}),
            ('Product_B', 200 + {date[-2:]})
        ) AS t(product, revenue)
    """)

# 定义宏：接受表名作为参数
conn.execute("""
    CREATE OR REPLACE MACRO get_daily_sales(tbl TABLE) AS TABLE
        SELECT product, SUM(revenue) AS total_revenue
        FROM tbl
        GROUP BY product;
""")

# 动态查询不同日期的表
for date in ['20231001', '20231002']:
    result = conn.execute(f"""
        SELECT * FROM get_daily_sales(sales_{date})
        ORDER BY product
    """).fetchdf()
    print(f"--- {date} ---")
    print(result)

import duckdb

conn = duckdb.connect()
conn.execute("""
    CREATE TABLE raw_data AS
    SELECT * FROM (VALUES
        ('  Active  ', '2023/10/01', 'NY'),
        ('inactive', '10-02-2023', 'ca'),
        ('PENDING', '2023.10.03', 'TX')
    ) AS t(status, date_str, state)
""")

# 宏1：标准化状态码
conn.execute("""
    CREATE OR REPLACE MACRO clean_status(s) AS
        CASE
            WHEN LOWER(TRIM(s)) IN ('active', 'act') THEN 'ACTIVE'
            WHEN LOWER(TRIM(s)) IN ('inactive', 'inact') THEN 'INACTIVE'
            ELSE UPPER(TRIM(s))
        END;
""")

# 宏2：标准化日期（兼容多种分隔符）
conn.execute("""
    CREATE OR REPLACE MACRO clean_date(d) AS
        strptime(REGEXP_REPLACE(d, '[./]', '-'), '%Y-%m-%d');
""")

# 执行清洗
result = conn.execute("""
    SELECT 
        clean_status(status) AS clean_status,
        clean_date(date_str) AS clean_date,
        UPPER(state) AS state
    FROM raw_data
""").fetchdf()

print(result)

import duckdb

conn = duckdb.connect()
conn.execute("""
    CREATE TABLE user_events AS
    SELECT * FROM (VALUES
        (1, '[{"type":"view","value":5},{"type":"click","value":3}]'),
        (2, '[{"type":"click","value":2},{"type":"purchase","value":100}]'),
        (3, '[{"type":"view","value":1}]')
    ) AS t(user_id, events_json)
""")

# 定义宏：提取特定类型事件的 value 总和
conn.execute("""
    CREATE OR REPLACE MACRO sum_event_value(events_json, event_type) AS
        (
            SELECT COALESCE(SUM(e.value), 0)
            FROM json_each(events_json) AS je
            CROSS JOIN LATERAL (
                SELECT json_extract_string(je.value, '$.type') AS type,
                       json_extract_int(je.value, '$.value') AS value
            ) AS e
            WHERE e.type = event_type
        );
""")

# 查询每个用户的点击总价值
result = conn.execute("""
    SELECT 
        user_id,
        sum_event_value(events_json, 'click') AS total_click_value,
        sum_event_value(events_json, 'view') AS total_view_value
    FROM user_events
    ORDER BY user_id
""").fetchdf()

print(result)

import duckdb

conn = duckdb.connect()
conn.execute("""
    CREATE TABLE customers AS
    SELECT * FROM (VALUES
        (1, 5, 300.0),
        (2, 10, 1500.0),
        (3, 2, 80.0)
    ) AS t(cust_id, order_count, total_spent)
""")

# 定义宏：根据消费行为返回客户等级和标签（多列）
conn.execute("""
    CREATE OR REPLACE MACRO customer_segment(count, spent) AS TABLE
    SELECT 
        CASE 
            WHEN spent >= 1000 THEN 'VIP'
            WHEN spent >= 100 THEN 'Standard'
            ELSE 'New'
        END AS tier,
        CASE 
            WHEN count >= 10 THEN 'Frequent'
            ELSE 'Occasional'
        END AS frequency_label
    WHERE 1=1;
""")

# 使用宏并展开多列
result = conn.execute("""
    SELECT 
        cust_id,
        cs.tier,
        cs.frequency_label
    FROM customers,
    LATERAL customer_segment(order_count, total_spent) AS cs
    ORDER BY cust_id
""").fetchdf()

print(result)

import duckdb

conn = duckdb.connect()
conn.execute("""
    CREATE TABLE sales AS
    SELECT * FROM (VALUES
        ('2023-10-01', 'North', 'Electronics', 1000.0),
        ('2023-10-01', 'South', 'Clothing', 500.0),
        ('2023-10-02', 'North', 'Electronics', 1500.0),
        ('2023-10-02', 'South', 'Clothing', 800.0)
    ) AS t(sale_date, region, category, amount)
""")

# 定义宏：计算某指标占总销售额比例
conn.execute("""
    CREATE OR REPLACE MACRO pct_of_total(part, total) AS
        CASE 
            WHEN total > 0 THEN ROUND(100.0 * part / total, 2)
            ELSE 0 
        END;
""")

# 动态生成多维度统计 SQL
dimensions = ['region', 'category']
base_query = """
    SELECT 
        '{dim}' AS dimension_type,
        {dim} AS dimension_value,
        SUM(amount) AS total_amount,
        pct_of_total(SUM(amount), (SELECT SUM(amount) FROM sales)) AS pct_total
    FROM sales
    GROUP BY {dim}
"""

# 拼接所有维度查询并执行
all_queries = " UNION ALL ".join([
    base_query.format(dim=d) for d in dimensions
])

result = conn.execute(all_queries).fetchdf()
print(result)