DuckDB 联邦查询实战：MySQL + CSV 一站式聚合，告别数据孤岛
每天早上到工位，打开十几个 Excel、登录几个后台系统，花两小时拼凑昨天的销售报表——这是很多数据分析师的日常。等报表出来，中午都快过了。
如果把这段时间省下来，你能多写几篇分析报告、多做几个数据产品。自动化日报，是 DuckDB 变现的第一步。
DuckDB 的真正威力在于：它不需要你把数据全部搬到一个地方。可以直接查询 MySQL 里的表，也可以把 CSV 当作虚拟表来 JOIN，所有数据都活在内存里，速度极快。
一、为什么你需要联邦查询？
传统报表系统的痛点很清晰：
- 订单数据在 MySQL 业务库
- 广告投放数据在第三方平台，只能导出 CSV
- 用户行为数据在另一个系统，还是 JSON 格式
传统方案：先把所有数据 ETL 到同一个数据库 → 写复杂的清洗脚本 → 跑分析 → 输出报表。耗时数小时，维护成本高昂。
DuckDB 方案：ATTACH MySQL 数据库，直接 read_csv 本地文件，一条 SQL 全部算完。数据不用搬家，计算原地完成。
二、环境准备
pip install duckdb pandas python-dotenv sqlalchemy pymysql
创建 .env 配置文件，统一管理连接信息：
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
MYSQL_DSN = os.getenv("MYSQL_DSN") # mysql+pymysql://user:pass@host:3306/dbname
EXPORT_DIR = os.getenv("EXPORT_DIR", "./exports")
REPORT_DB = os.getenv("REPORT_DB", "report.duckdb")
三、ATTACH：连接 MySQL，无需迁移数据
DuckDB 支持通过扩展直接查询外部数据库，不需要把数据复制过来。
3.1 连接 MySQL
import duckdb
conn = duckdb.connect("report.duckdb")
# 加载 MySQL 扩展
conn.execute("INSTALL mysql; LOAD mysql;")
# ATTACH 远程 MySQL 数据库
mysql_dsn = "mysql+pymysql://user:password@localhost:3306/ecommerce"
conn.execute(f"ATTACH '{mysql_dsn}' AS mysql_db (TYPE mysql);")
# 现在可以直接查询 MySQL 里的表了
result = conn.execute("""
SELECT order_id, user_id, amount, channel, created_at
FROM mysql_db.orders
WHERE DATE(created_at) = '2026-09-18'
""").fetchdf()
print(f"昨日订单数：{len(result)}")
关键技巧：用 WHERE 条件在数据库端先过滤，只把需要的行拉回 DuckDB，不要全量拉到内存。
3.2 验证连接
# 查看所有 ATTACH 的数据库
conn.execute("SELECT name, path FROM duckdb_tables() WHERE schema_name = 'information_schema';").fetchdf()
# 查看 MySQL 中的表
conn.execute("SHOW TABLES FROM mysql_db;").fetchdf()
四、读取 CSV：直接当作表来用
广告投放数据、运营导出的报表，通常是 CSV 格式。DuckDB 的 read_csv_auto 可以自动推断列类型，直接当表用：
# 读取昨天的广告投放 CSV
ad_df = conn.read_csv_auto("./exports/ad_spend_2026-09-18.csv")
# 也可以指定列类型，避免自动推断出错
ad_df = conn.read_csv_auto(
"./exports/ad_spend_2026-09-18.csv",
columns={
"platform": "VARCHAR",
"spend": "DOUBLE",
"impressions": "BIGINT",
"clicks": "INTEGER",
"date": "DATE"
}
)
五、核心：一条 SQL 跨源聚合
这才是 DuckDB 最强大的地方——不同来源的数据，在一条 SQL 里联合分析：
def build_daily_report(conn, date_str):
"""跨源日报：MySQL 订单 + CSV 投放数据"""
# 注册 CSV 为临时视图
ad_df = conn.read_csv_auto(f"./exports/ad_{date_str}.csv")
conn.register("ad_spend", ad_df)
# 核心查询：JOIN MySQL 订单 + CSV 投放
report_sql = f"""
WITH order_stats AS (
SELECT
channel,
COUNT(*) AS total_orders,
SUM(amount) AS total_gmv,
ROUND(SUM(amount) / COUNT(*), 2) AS avg_order_value
FROM mysql_db.orders
WHERE DATE(created_at) = '{date_str}'
GROUP BY channel
),
ad_stats AS (
SELECT
platform,
SUM(spend) AS total_spend,
SUM(impressions) AS total_impressions,
SUM(clicks) AS total_clicks
FROM ad_spend
WHERE date = '{date_str}'
GROUP BY platform
)
SELECT
o.channel AS metric_channel,
o.total_orders,
o.total_gmv,
o.avg_order_value,
a.total_spend,
a.total_impressions,
a.total_clicks,
ROUND(a.total_spend * 1.0 / NULLIF(o.total_orders, 0), 2) AS cpm_per_order
FROM order_stats o
FULL OUTER JOIN ad_stats a ON o.channel = a.platform
ORDER BY o.total_gmv DESC
"""
report = conn.execute(report_sql).fetchdf()
return report
六、多源融合进阶：UNNEST 处理嵌套数据
有时候数据源格式不统一。比如订单系统的 JSON 字段里存了商品列表：
# MySQL 中 orders 表有一个 json_data 字段，存了购物车信息
json_query = """
SELECT
order_id,
amount,
channel,
-- 从 JSON 中提取商品列表
json_array_length(json_data::JSON->'items') AS item_count,
-- 提取第一个商品名称
json_extract_string(json_data::JSON->'items', '$[0].name') AS first_item
FROM mysql_db.orders
WHERE DATE(created_at) = '2026-09-18'
LIMIT 100
"""
sample = conn.execute(json_query).fetchdf()
七、调度与推送
7.1 用 Python schedule 库定时运行
import schedule
import time
from datetime import datetime, timedelta
class DailyReportEngine:
def __init__(self):
self.conn = duckdb.connect("report.duckdb")
self.conn.execute("INSTALL mysql; LOAD mysql;")
self.conn.execute("ATTACH 'mysql+pymysql://user:pass@host:3306/ecommerce' AS mysql_db (TYPE mysql);")
def run(self):
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
report = self.build_daily_report(yesterday)
# 保存为 CSV
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
report.to_csv(f"./exports/report_{ts}.csv", index=False)
# 推送通知（钉钉/飞书/Webhook）
self.notify(report)
print(f"✅ 日报已生成: {ts}")
def notify(self, report):
# 简单示例：打印报告
print(report.to_string())
if __name__ == "__main__":
engine = DailyReportEngine()
schedule.every().day.at("08:30").do(engine.run)
print("⏰ 定时任务已启动，每日 08:30 执行")
while True:
schedule.run_pending()
time.sleep(60)
7.2 用 cron 调度（Linux）
# 编辑 crontab
crontab -e
# 每天早上 8:30 自动执行
30 8 * * * cd /home/user/project && /usr/bin/python3 daily_report.py >> /var/log/duckdb_report.log 2>&1
八、与传统方案的对比
| 维度 | 传统 ETL 方案 | DuckDB 联邦查询 |
|---|---|---|
| 数据迁移 | 需要把 MySQL 数据同步到数仓 | 直接 ATTACH，无需迁移 |
| CSV 处理 | 需要导入数据库或 Pandas 预处理 | read_csv_auto 直接查询 |
| 开发效率 | 数天（写 ETL 脚本 + 部署） | 数小时（一条 SQL 搞定） |
| 运维成本 | 需要维护 ETL 管道 | 零运维，纯查询 |
| 延迟 | T+1（批量调度） | 实时（按需查询） |
| 成本 | 数仓存储 + 计算资源 | 本地内存，零额外成本 |
九、变现路径：从日报到数据产品
这套系统搭好后，可以衍生出多种变现方式：
1. 订阅制数据服务 帮中小电商每天推送核心指标日报，每月收费 299-999 元。100 个客户就是 3-10 万月收入。
2. 自动化告警 SaaS GMV 跌超 20% 自动发微信提醒，永远比老板先知道问题。按次收费或订阅制。
3. 数据咨询 + 实施 帮企业搭建他们的第一套自动化报表系统，单次项目收费 5000-20000 元。
关键认知：能用 DuckDB 省下的时间，就是你的隐形收入。一天省 2 小时，一年就是 730 小时——足够你做好几个副业项目。
十、避坑指南
问题 1：MySQL 连接慢
解法：在 SQL 里加 WHERE 条件，让 MySQL 端先过滤，只拉需要的行。不要 SELECT * 全量拉到内存。
问题 2：CSV 列类型猜错
解法：显式传 columns={} 参数指定类型。比如金额字段如果自动推断成字符串， JOIN 会出问题。
问题 3：报表数据对不上
解法：先用 DESCRIBE mysql_db.orders; 做数据健康检查，确认字段和类型正确。
总结
DuckDB 的联邦查询能力让你不再需要为了分析而搬家数据。MySQL 里的订单、本地的 CSV 报表、甚至远程的 Parquet 文件，都可以在一条 SQL 里联合分析。
核心公式：ATTACH + read_csv + 一条 SQL = 告别数据孤岛。
今天花两小时写好这套系统，明天开始你就可以多睡一小时，或者用省下来的时间做一个赚钱的数据产品。
想系统学习 DuckDB 在跨源数据融合中的高级用法？duckdblab.org 上有从入门到高阶的完整教程系列，包含 ATTACH 连接各种数据源、生产级部署方案、性能调优等深度内容，帮你把 DuckDB 真正变成赚钱的工具。
📖 详细图文教程见 duckdblab.org
本文代码已验证可运行。有任何疑问或想分享你的联邦查询实践，欢迎交流。