import duckdb

con = duckdb.connect("analytics.duckdb")

# 方式 1：从 DataFrame 直接写入（最常用）
import pandas as pd
df = pd.read_csv("large_dataset.csv")
con.execute("CREATE TABLE events AS SELECT * FROM df")
con.execute("COPY events TO 'events.parquet' (FORMAT PARQUET)")

# 方式 2：一行命令从 CSV 直接转 Parquet
con.execute("""
    COPY (SELECT * FROM read_csv_auto('large_dataset.csv'))
    TO 'events.parquet' (FORMAT PARQUET)
""")

# 方式 3：多文件并行写入（利用多线程）
con.execute("""
    COPY (SELECT * FROM read_csv_auto('data/*.csv'))
    TO 'events_output.parquet' (FORMAT PARQUET, COMPRESSION ZSTD, CHUNK_SIZE 32768)
""")
print("✅ Parquet 文件已生成")

-- 不同压缩算法的性能与大小对比
-- ZSTD: 最佳压缩率，适合存储
-- SNAPPY: 平衡速度和压缩率，推荐默认
-- UNCOMPRESSED: 最大速度，适合临时文件

-- 使用 ZSTD 压缩（压缩比最高，适合长期存储）
COPY events TO 'events_zstd.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);

-- 使用 SNAPPY 压缩（平衡方案，推荐生产使用）
COPY events TO 'events_snappy.parquet' (FORMAT PARQUET, COMPRESSION SNAPPY);

-- 使用 BROTLI 压缩（压缩率接近 ZSTD，解压更快）
COPY events TO 'events_brotli.parquet' (FORMAT PARQUET, COMPRESSION BROTLI);

-- 查看不同压缩方式的文件大小
SELECT 
    filename,
    round(file_size / 1024 / 1024, 2) AS size_mb,
    round(file_size / 1024 / 1024 / 1024, 2) AS size_gb
FROM (VALUES
    ('events_uncompressed.parquet', system.parquet_metadata('events_uncompressed.parquet').row_groups[1].total_compressed_size),
    ('events_snappy.parquet', system.parquet_metadata('events_snappy.parquet').row_groups[1].total_compressed_size),
    ('events_zstd.parquet', system.parquet_metadata('events_zstd.parquet').row_groups[1].total_compressed_size)
) AS t(filename, file_size);

import duckdb

con = duckdb.connect("analytics.duckdb")

# ❌ 错误做法：读取所有列（浪费 IO 和内存）
df_bad = con.execute("SELECT * FROM events.parquet").df()

# ✅ 正确做法：只读需要的列（减少 80%+ IO）
df_good = con.execute("""
    SELECT event_type, user_id, timestamp 
    FROM read_parquet('events.parquet')
    WHERE timestamp >= '2026-01-01'
""").df()

# ✅ 进阶：利用谓词下推（Filter Pushdown）
# DuckDB 会自动把 WHERE 条件推到 Parquet 读取层
df_optimized = con.execute("""
    SELECT user_id, event_type, COUNT(*) as cnt
    FROM read_parquet('events.parquet')
    WHERE event_type IN ('purchase', 'signup')
      AND timestamp >= '2026-06-01'
    GROUP BY user_id, event_type
""").df()

-- DuckDB 会自动优化为：
-- 1. 读取 Parquet 文件的统计信息（min/max of each column）
-- 2. 跳过不满足条件的 Row Group
-- 3. 只读取满足条件的数据块

-- 示例：10GB Parquet 文件，只读 1% 的数据
EXPLAIN ANALYZE
SELECT *
FROM read_parquet('events_large.parquet')
WHERE event_date = '2026-09-25'
  AND user_id > 1000000;

-- 输出示例（关键信息）：
-- Output: []
--   Parallel Project: []
--     Parallel TableFunction: read_parquet
--       Filters: event_date = '2026-09-25' AND user_id > 1000000
--       RowGroupsSkipped: 48/50  ← 跳过了 96% 的数据！
--       RowsRead: 2,345,678 / 234,567,890  ← 只读了 1%

-- 创建分区目录结构
-- data/
-- ├── year=2024/month=01/events.parquet
-- ├── year=2024/month=02/events.parquet
-- ├── ...
-- └── year=2026/month=09/events.parquet

-- 自动识别 Hive 风格分区
SELECT *
FROM read_parquet('data/year=*/month=*/events.parquet')
WHERE year = 2026 AND month = 9;

-- DuckDB 只会读取 2026-09 这一个分区文件
-- 而不是扫描全部 33 个月的数据！

import duckdb

con = duckdb.connect("analytics.duckdb")

# 场景：每个月一个 Parquet 文件，需要跨月聚合
# 自动识别目录下的所有 Parquet 文件
result = con.execute("""
    SELECT 
        strftime(timestamp, '%Y-%m') AS month,
        event_type,
        COUNT(*) AS event_count,
        SUM(amount) AS total_amount
    FROM read_parquet('data/monthly_*.parquet', union_by_name=true)
    WHERE timestamp >= '2026-01-01'
    GROUP BY month, event_type
    ORDER BY month DESC
""").df()

print(result.to_markdown())

-- 查看 Parquet 文件的列编码信息
SELECT 
    column_name,
    compression,
    num_values,
    num_rows,
    null_count,
    distinct_count,
    round(compressed_size / 1024 / 1024, 2) AS compressed_mb,
    round(uncompressed_size / 1024 / 1024, 2) AS uncompressed_mb
FROM parquet_schema('events.parquet')
ORDER BY column_name;

import duckdb

con = duckdb.connect("analytics.duckdb")

# 大文件分块处理，避免内存溢出
chunk_size = 100_000  # 每次处理 10 万行

for i, batch in enumerate(con.execute("""
    SELECT * FROM read_parquet('huge_dataset.parquet')
""").fetchmany(chunk_size)):
    # 处理每个批次
    process_batch(batch)
    print(f"Processed batch {i+1}")

import duckdb
import pyarrow as pa

con = duckdb.connect("analytics.duckdb")

# DuckDB 可以直接输出 Arrow 格式，无需复制数据
arrow_table = con.execute("SELECT * FROM read_parquet('events.parquet')").arrow()

# 直接传给 Polars / Pandas，零拷贝！
import polars as pl
df = pl.from_arrow(arrow_table)

# 或者传给机器学习模型
from sklearn.ensemble import RandomForestClassifier
X = arrow_table.to_pandas()[['feature1', 'feature2', 'feature3']]
y = arrow_table.to_pandas()['label']

-- DuckDB 自动并行读取多个 Parquet 文件
-- 利用所有 CPU 核心加速
SELECT 
    region,
    SUM(sales) AS total_sales,
    AVG(price) AS avg_price
FROM read_parquet('/data/sales/*.parquet')
GROUP BY region;

-- 你可以控制并发度
SET parallel_threads = 8;  -- 默认使用所有核心

-- 1. 使用合适的压缩算法
-- 频繁读取 → SNAPPY（平衡）
-- 长期存储 → ZSTD（压缩比高）
-- 临时数据 → UNCOMPRESSED（最快）

-- 2. 合理设置 chunk 大小
COPY large_table TO 'output.parquet' 
(FORMAT PARQUET, COMPRESSION ZSTD, CHUNK_SIZE 65536);

-- 3. 分区写入
COPY (SELECT * FROM source WHERE date >= '2026-01-01')
TO 'output_by_month/' (FORMAT PARQUET, PARTITION_BY (date_trunc('month', date)));

-- 1. 始终指定需要的列，不要用 SELECT *
SELECT user_id, event_type, timestamp FROM events.parquet;

-- 2. 利用 WHERE 条件触发谓词下推
SELECT user_id FROM events.parquet 
WHERE event_type = 'purchase' AND timestamp > '2026-09-01';

-- 3. 对大表使用 LIMIT 预览
SELECT * FROM read_parquet('huge.parquet') LIMIT 100;

-- 4. 检查执行计划确认优化生效
EXPLAIN ANALYZE 
SELECT user_id, COUNT(*) FROM read_parquet('events.parquet')
WHERE event_date = '2026-09-25' GROUP BY user_id;