DuckDB JSON 数据处理实战工作坊：嵌套解析、数组展开与生产应用

-- 读取单个 JSON 文件
SELECT * FROM read_json_auto('users.json');

-- 读取目录下所有 JSON 文件  
SELECT * FROM read_json_auto('/data/*.json');

-- 递归读取子目录
SELECT * FROM read_json_auto('/data/**/*.json', recursive=true);

{
    "order_id": "ORD-2024-001",
    "customer": {
        "name": "张三",
        "tier": "VIP"
    },
    "items": [
        {"product": "笔记本电脑", "price": 5999, "qty": 1},
        {"product": "鼠标", "price": 199, "qty": 2}
    ],
    "total": 6397,
    "timestamp": "2024-01-15T10:30:00Z"
}

CREATE TABLE orders AS
SELECT 
    o.order_id,
    o.customer->>'name' AS customer_name,
    o.customer->>'tier' AS customer_tier,
    o.total,
    o.timestamp
FROM read_json_auto('orders.json') o;

-- 查看结果
SELECT * FROM orders LIMIT 5;

-- 将 items 数组展开为一行一条商品记录
SELECT 
    order_id,
    item->>'product' AS product_name,
    (item->>'price')::INTEGER AS price,
    (item->>'qty')::INTEGER AS quantity,
    (item->>'price')::INTEGER * (item->>'qty')::INTEGER AS line_total
FROM read_json_auto('orders.json'),
     UNNEST(items) AS item;

SELECT 
    log_timestamp,
    json_extract_path(json_data, 'user', 'id') AS user_id,
    json_extract_path(json_data, 'user', 'name') AS user_name,
    json_extract_path(json_data, 'payload', 'page') AS page,
    json_extract_path(json_data, 'payload', 'amount') AS amount
FROM api_logs;

CREATE TABLE user_events AS
SELECT * FROM VALUES
    (1, '[{"type": "click", "time": "10:00"}, {"type": "scroll", "time": "10:02"}]'),
    (2, '[{"type": "purchase", "time": "11:00"}]')
AS user_events(user_id, events_json);

-- 把数组展开成多行
SELECT 
    user_id,
    (event ->> 'type') AS event_type,
    event ->> 'time' AS event_time
FROM user_events
CROSS JOIN LATERAL json_array_elements(events_json) AS event;

-- 按用户聚合其所有订单为 JSON 数组
SELECT 
    customer_name,
    json_group_array(ORDER_OBJECT) AS order_history
FROM (
    SELECT 
        customer_name,
        json_build_object(
            'order_id', order_id,
            'total', total,
            'date', timestamp
        ) AS ORDER_OBJECT
    FROM orders
) GROUP BY customer_name;

SELECT 
    page_path AS page,
    COUNT(*) AS visit_count,
    SUM(COALESCE(payload_duration::INT, 0)) AS total_duration_seconds
FROM (
    SELECT 
        log_timestamp,
        json_data -> 'payload' ->> 'page' AS page_path,
        json_data -> 'payload' ->> 'duration' AS payload_duration
    FROM api_logs
) sub
WHERE page_path IS NOT NULL
GROUP BY page_path
ORDER BY visit_count DESC;

UPDATE api_logs
SET json_data = json_set(json_data, 'payload', 
    json_set(json_data -> 'payload', 'duration', '60'))
WHERE json_data ->> 'action' = 'click'
  AND json_extract_path(json_data, 'payload', 'duration') IS NULL;

# Pandas 方式
import pandas as pd
import json

with open('events.json', 'r') as f:
    data = json.load(f)
df = pd.json_normalize(data)

# DuckDB 方式
import duckdb
con = duckdb.connect()
df = con.execute("SELECT * FROM read_json_auto('events.json')").fetchdf()

CREATE MATERIALIZED VIEW mv_api_logs AS
SELECT 
    log_timestamp,
    json_data ->> 'action' AS action,
    json_data -> 'payload' ->> 'page' AS page,
    json_data -> 'payload' ->> 'duration' AS duration
FROM api_logs;

-- 之后在 MV 上建索引，查询更快
CREATE INDEX idx_mv_page ON mv_api_logs(page);

SELECT * FROM read_csv('logs/*.json', format='json');

-- 导出为 Parquet（压缩率更高，后续查询更快）
COPY (SELECT * FROM orders) TO '/data/orders.parquet' (FORMAT PARQUET);

-- 导出为 CSV
COPY (SELECT * FROM orders) TO '/data/orders.csv' (FORMAT CSV);

-- 导出为 JSON
COPY (SELECT * FROM orders) TO '/data/orders.json' (FORMAT JSON);

from fastapi import FastAPI
import duckdb

app = FastAPI()

@app.get("/api/analytics")
def get_analytics():
    con = duckdb.connect(":memory:")
    result = con.execute("""
        SELECT 
            json_extract_scalar(event, '$.type') AS event_type,
            COUNT(*) AS count
        FROM read_json_auto('https://api.example.com/events.json')
        GROUP BY event_type
    """).fetchdf()
    return result.to_dict()

# daily_report.py
import duckdb
import requests
from datetime import datetime

con = duckdb.connect()

# 1. 获取当日数据
resp = requests.get("https://api.example.com/today/data")
data = resp.json()

# 2. 写入临时表
con.execute("CREATE TEMP TABLE raw_data (payload JSONB)")
con.execute("INSERT INTO raw_data VALUES (?)", [data])

# 3. 分析处理
result = con.execute("""
    SELECT 
        country,
        COUNT(*) as visits,
        AVG(price) as avg_order_value
    FROM raw_data, UNNEST(payload.users) AS u
    GROUP BY country
""").fetchdf()

# 4. 保存结果
con.execute("COPY result TO '/reports/daily_{}.csv'".format(datetime.now().strftime('%Y%m%d')))