你有没有遇到过这种场景：手里有一份几万行的 CSV 销售数据，想用 SQL 做聚合查询，但 Excel 打开就卡死，导入数据库又嫌麻烦。
今天做一个本地数据分析工作台。一个 Python 脚本，直接对 CSV 文件跑 SQL 查询，支持聚合统计、条件筛选、分组汇总，结果一键导出。百万行数据，几秒钟出结果。
项目背景
传统的数据分析流程是：导数据 → 建数据库 → 写查询 → 导结果。每一步都要配置，小项目根本不值得这么折腾。
DuckDB 解决了这个问题。它是一个嵌入式的分析型数据库，不需要安装服务器，不需要配置连接。直接对 CSV 文件执行 SQL，速度比 pandas 快好几倍。
适用场景：
- 对 CSV/JSON 文件做交互式查询
- 临时分析导出数据，不想折腾数据库
- 跑聚合统计和交叉分析
- 把多张 CSV 合并成一张汇总表
技术选型
| 组件 | 选择 | 理由 |
|---|---|---|
| 查询引擎 | DuckDB | 嵌入式 SQL 引擎，零配置，CSV 直读 |
| 交互界面 | Rich + 命令行 | 彩色表格输出，终端体验好 |
| 数据导出 | pandas | 灵活的 CSV/Excel 导出 |
| 文件浏览 | pathlib | 自动扫描 CSV 文件 |
不用装服务器，不用配置连接池。一个 pip install 搞定所有依赖。
实现步骤
第一步：安装依赖
pip install duckdb rich pandas openpyxl
就四个包。duckdb 是核心，rich 负责终端输出美化，pandas 用于导出。
第二步：创建分析引擎 analyzer.py
"""
analyzer.py - DuckDB 本地数据分析引擎
直接对 CSV 文件执行 SQL，支持查询、统计、导出
"""
import duckdb
import pandas as pd
from pathlib import Path
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
console = Console()
class DataAnalyzer:
"""本地 CSV 数据分析器"""
def __init__(self, data_dir="."):
self.data_dir = Path(data_dir)
self.conn = duckdb.connect(":memory:")
self._registered_tables = {}
def scan_csv_files(self):
"""扫描目录下的所有 CSV 文件"""
csv_files = list(self.data_dir.glob("*.csv"))
if not csv_files:
console.print("[red]没有找到 CSV 文件[/red]")
return []
console.print(f"\n[bold cyan]发现 {len(csv_files)} 个 CSV 文件：[/bold cyan]")
for f in csv_files:
rows = self._quick_count(f)
console.print(f" • {f.name} — {rows} 行")
return csv_files
def _quick_count(self, filepath):
"""快速统计行数"""
return self.conn.execute(
f"SELECT COUNT(*) FROM read_csv_auto('{filepath}')"
).fetchone()[0]
def register(self, csv_path, table_name=None):
"""注册 CSV 文件为 DuckDB 表"""
path = Path(csv_path)
if not path.exists():
console.print(f"[red]文件不存在：{path}[/red]")
return None
if table_name is None:
table_name = path.stem
self.conn.execute(f"DROP TABLE IF EXISTS {table_name}")
self.conn.execute(
f"CREATE TABLE {table_name} AS SELECT * FROM read_csv_auto('{path}')"
)
self._registered_tables[table_name] = str(path)
rows = self.conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
cols = self.conn.execute(f"DESCRIBE {table_name}").fetchall()
console.print(f"\n[bold green]✓ 已加载：{table_name}[/bold green]")
console.print(f" 文件：{path.name}")
console.print(f" 行数：{rows:,}")
console.print(f" 列数：{len(cols)}")
# 显示列信息
col_table = Table(title="字段列表", show_header=False, box=None)
for name, dtype in cols:
col_table.add_row(name, dtype)
console.print(col_table)
return table_name
def query(self, sql, table_name=None):
"""执行 SQL 查询，返回结果"""
try:
result = self.conn.sql(sql)
df = result.fetchdf()
return df
except Exception as e:
console.print(f"[red]查询出错：{e}[/red]")
return None
def show(self, table_name, limit=20):
"""预览表数据"""
df = self.query(f"SELECT * FROM {table_name} LIMIT {limit}")
if df is None:
return
# 用 rich 表格输出
table = Table(title=f"{table_name} — 前 {limit} 行")
for col in df.columns:
table.add_column(str(col))
for _, row in df.iterrows():
table.add_row(*[str(v) if pd.notna(v) else "" for v in row])
console.print(table)
def describe(self, table_name):
"""显示表的基本统计信息"""
df = self.query(f"DESCRIBE {table_name}")
if df is None:
return
table = Table(title=f"{table_name} — 字段信息")
table.add_column("字段名")
table.add_column("类型")
table.add_column("null_count")
table.add_column("distinct_count")
for _, row in df.iterrows():
name = row["column_name"]
dtype = row["column_type"]
table.add_row(str(name), str(dtype), "", "")
console.print(table)
def summarize(self, table_name, group_col, value_col, agg="SUM"):
"""分组聚合统计"""
agg_map = {
"SUM": "SUM", "AVG": "AVG", "COUNT": "COUNT",
"MAX": "MAX", "MIN": "MIN", "MEDIAN": "MEDIAN"
}
agg_fn = agg_map.get(agg.upper(), "SUM")
sql = f"""
SELECT "{group_col}", {agg_fn}("{value_col}") as result
FROM {table_name}
GROUP BY "{group_col}"
ORDER BY result DESC
LIMIT 20
"""
df = self.query(sql)
if df is None:
return df
table = Table(title=f"{table_name} — {agg} by {group_col}")
table.add_column(group_col)
table.add_column(f"{agg}({value_col})", justify="right")
for _, row in df.iterrows():
val = row["result"]
if isinstance(val, float):
val = f"{val:,.2f}"
table.add_row(str(row[group_col]), str(val))
console.print(table)
return df
def export(self, df, output_path):
"""导出数据到 CSV 或 Excel"""
path = Path(output_path)
if path.suffix in (".csv", ".tsv"):
df.to_csv(path, index=False, encoding="utf-8-sig")
elif path.suffix in (".xlsx", ".xls"):
df.to_excel(path, index=False)
else:
df.to_csv(path, index=False, encoding="utf-8-sig")
console.print(f"\n[bold green]✓ 已导出：{path.absolute()}[/bold green]")
console.print(f" 共 {len(df):,} 行，{len(df.columns)} 列")
return path
核心就一个类：DataAnalyzer。注册 CSV 文件 → 执行 SQL → 导出结果。全程不需要任何数据库配置。
第三步：创建交互命令行 tool.py
"""
tool.py - 交互式数据分析工具
运行方式：python tool.py
"""
import sys
import argparse
from pathlib import Path
from analyzer import DataAnalyzer
console = DataAnalyzer()
def cmd_list(args):
"""列出当前目录的 CSV 文件"""
console.data_dir = Path(args.dir or ".")
console.scan_csv_files()
def cmd_register(args):
"""注册 CSV 文件"""
console.register(args.file, args.name)
def cmd_query(args):
"""执行自定义 SQL"""
table = args.table
sql = args.sql
# 如果 SQL 里没有 FROM，自动加上表名
if "FROM" not in sql.upper() and table:
sql = sql.replace("SELECT", f"SELECT ", 1)
sql = f"SELECT * FROM {table} WHERE {sql}" if table else sql
df = console.query(sql, table)
if df is not None:
if args.limit:
df = df.head(args.limit)
console.show_df(df, title="查询结果")
if args.output:
console.export(df, args.output)
def cmd_stats(args):
"""分组统计"""
df = console.summarize(args.table, args.group, args.value, args.agg)
if df is not None and args.output:
console.export(df, args.output)
def cmd_describe(args):
"""查看表结构"""
console.describe(args.table)
console.show(args.table, limit=10)
def cmd_df(df, title="结果"):
"""用 rich 表格打印 DataFrame"""
from rich.table import Table
table = Table(title=title)
for col in df.columns:
table.add_column(str(col))
for _, row in df.iterrows():
table.add_row(*[str(v) if __import__('pandas').isna(v) is False else "" for v in row])
console.print(table)
def main():
parser = argparse.ArgumentParser(
description="DuckDB 本地数据分析工具",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例：
python tool.py list # 列出 CSV 文件
python tool.py register data.csv # 注册数据文件
python tool.py query sales "amount > 1000" --table sales
python tool.py stats sales --group region --value amount --agg SUM
python tool.py describe sales
"""
)
sub = parser.add_subparsers(dest="command", required=True)
# list
p_list = sub.add_parser("list", help="列出 CSV 文件")
p_list.add_argument("--dir", "-d", default=".", help="扫描目录")
# register
p_reg = sub.add_parser("register", help="注册 CSV 文件")
p_reg.add_argument("file", help="CSV 文件路径")
p_reg.add_argument("--name", "-n", help="表名（默认用文件名）")
# query
p_q = sub.add_parser("query", help="执行 SQL 查询")
p_q.add_argument("sql", nargs="?", help="SQL 语句或 WHERE 条件")
p_q.add_argument("--table", "-t", help="表名")
p_q.add_argument("--limit", "-l", type=int, help="限制行数")
p_q.add_argument("--output", "-o", help="导出文件路径")
# stats
p_s = sub.add_parser("stats", help="分组统计")
p_s.add_argument("--table", "-t", required=True, help="表名")
p_s.add_argument("--group", "-g", required=True, help="分组字段")
p_s.add_argument("--value", "-v", required=True, help="数值字段")
p_s.add_argument("--agg", "-a", default="SUM",
choices=["SUM", "AVG", "COUNT", "MAX", "MIN", "MEDIAN"])
p_s.add_argument("--output", "-o", help="导出文件路径")
# describe
p_d = sub.add_parser("describe", help="查看表结构")
p_d.add_argument("--table", "-t", required=True, help="表名")
args = parser.parse_args()
if args.command == "list":
cmd_list(args)
elif args.command == "register":
cmd_register(args)
elif args.command == "query":
cmd_query(args)
elif args.command == "stats":
cmd_stats(args)
elif args.command == "describe":
cmd_describe(args)
if __name__ == "__main__":
main()
命令行工具支持 5 个子命令：list、register、query、stats、describe。每个命令对应一个常用操作。
第四步：创建快速上手脚本 quickstart.py
"""
quickstart.py - 快速演示脚本
直接运行：python quickstart.py
"""
from analyzer import DataAnalyzer
import duckdb
# 创建一个模拟数据集
console = DataAnalyzer()
# 用 DuckDB 直接创建测试数据
console.conn.execute("""
CREATE TABLE sales AS SELECT * FROM (VALUES
('2024-01', '华东', '电子产品', 12500, 50),
('2024-01', '华北', '日用品', 8300, 120),
('2024-01', '华南', '电子产品', 15600, 30),
('2024-02', '华东', '日用品', 9200, 80),
('2024-02', '华北', '电子产品', 18700, 45),
('2024-02', '华南', '服装', 6500, 60),
('2024-03', '华东', '服装', 11200, 40),
('2024-03', '华北', '日用品', 7800, 90),
('2024-03', '华南', '电子产品', 22100, 25),
('2024-03', '华东', '电子产品', 16800, 35),
) AS t(month, region, category, amount, qty)
""")
print("=" * 50)
print(" DuckDB 本地数据分析演示")
print("=" * 50)
# 1. 查看表结构
print("\n[1] 表结构")
result = console.conn.execute("DESCRIBE sales").fetchdf()
print(result.to_string(index=False))
# 2. 简单查询
print("\n[2] 查询：金额大于 10000 的记录")
df = console.query("SELECT * FROM sales WHERE amount > 10000")
print(df.to_string(index=False))
# 3. 分组汇总
print("\n[3] 按地区汇总销售额")
df = console.summarize("sales", "region", "amount", "SUM")
# 4. 多维度聚合
print("\n[4] 按月份和品类汇总")
df = console.query("""
SELECT month, category, SUM(amount) as total, SUM(qty) as total_qty
FROM sales
GROUP BY month, category
ORDER BY month, total DESC
""")
print(df.to_string(index=False))
# 5. 导出结果
console.export(df, "sales_summary.csv")
print("\n✓ 结果已导出到 sales_summary.csv")
运行效果
跑 python quickstart.py，输出是这样的：
==================================================
DuckDB 本地数据分析演示
==================================================
[1] 表结构
month region category amount qty
datetime varchar varchar bigint int64
[2] 查询：金额大于 10000 的记录
month region category amount qty
2024-01-01 华南 电子产品 22100 25
2024-02-01 华北 电子产品 18700 45
...
[3] 按地区汇总销售额
region SUM(amount)
华北 36800.00
华南 43800.00
华东 49700.00
[4] 按月份和品类汇总
month category total total_qty
2024-01 电子产品 28100 80
...
✓ 结果已导出到 sales_summary.csv
如果是命令行模式，用法是这样的：
# 列出 CSV 文件
python tool.py list
# 注册数据文件
python tool.py register sales.csv -n sales
# 执行查询
python tool.py query "region = '华南' AND amount > 10000" -t sales
# 分组统计
python tool.py stats -t sales -g region -v amount -a SUM
# 导出结果
python tool.py stats -t sales -g region -v amount -a AVG -o result.xlsx
优化方向
这个版本已经能用了，想继续打磨也可以：
- Jupyter 扩展：写一个 IPython magic，在 Notebook 里直接
%%duckdb执行 SQL - Web 界面：用 Streamlit 包一层，拖拽上传 CSV，可视化出图
- 数据清洗：加缺失值填充、类型转换、重复行去重等预处理功能
- 多文件合并：支持扫描目录，自动合并多个 CSV 为一张大表
- 导出 Excel 图表：用 openpyxl 在导出的 Excel 里直接生成透视表图表
- SQL 历史记录：保存常用查询，方便重复执行
DuckDB 的核心优势就两点：快和简单。不用装服务器，不用配置连接，直接对文件跑 SQL。百万行数据随便查。
打开终端，python quickstart.py，你的第一个 DuckDB 数据分析工具就跑起来了。要不要把你的 CSV 文件丢进去试试？