🚀

用 Python + DuckDB 做本地 CSV 数据分析,5 分钟搞定周报

不用装数据库,不用写代码,几行 Python 就能用 SQL 分析你的 CSV 数据。周末整理报销单、做销售报表,统统搞定。

👁️ - 次阅读

你有没有这种经历:领导突然要一份数据报表,你把 Excel 打开,筛选、透视表、公式堆了一小时,最后发现还少了一列。

或者每个月都要整理一堆 CSV 文件——销售记录、用户数据、日志导出——翻来翻去烦得要死。

如果用 SQL 直接查,30 秒出结果,还不用打开 Excel。

这个项目用 Python + DuckDB,教你怎么在本地快速分析 CSV 数据。不用装数据库,不用写前端,一个脚本跑通。

项目背景

CSV 文件太常见了。几乎每个系统都能导出 CSV:电商后台、CRM、服务器日志、第三方 API 回调。

但处理 CSV 的方式很痛苦:

  • Excel 打开几万行就卡死
  • 写 Python 循环手动处理,代码又臭又长
  • 装 MySQL/PostgreSQL 做个小分析,太重了

DuckDB 就是为了解决这个问题而生的。它是一个嵌入式 SQL 数据库,可以直接读 CSV 文件,用 SQL 做分析,速度比 pandas 快,比 Excel 轻便。

这个项目用 DuckDB 做一个简单的 CSV 数据分析工具,你可以用它处理报销单、销售报表、用户数据等任何 CSV 文件。

技术选型

组件选型理由
语言Python 3.10+生态丰富,脚本上手快
数据库DuckDB嵌入式 SQL,直接读 CSV,无需安装服务
数据格式CSV通用导出格式,几乎所有系统都支持
输出终端 / HTML 报告简单直接,方便查看

不需要装 PostgreSQL、MySQL,也不需要 pandas(虽然能用)。DuckDB 直接读 CSV,SQL 语法你大概也懂一点,上手很快。

实现步骤

第一步:安装 DuckDB

pip install duckdb

就这一行。DuckDB 是纯 Python 包,安装完直接用。

第二步:准备数据

随便找一个 CSV 文件来测试。我们用这个示例数据:

order_id,date,customer,product,quantity,price,status
1001,2026-07-01,张三,键盘,2,299,已完成
1002,2026-07-01,李四,鼠标,1,89,已完成
1003,2026-07-02,张三,显示器,1,1299,已完成
1004,2026-07-02,王五,键盘,1,299,已取消
1005,2026-07-03,李四,耳机,1,399,已完成
1006,2026-07-03,赵六,鼠标,3,89,已完成
1007,2026-07-04,张三,键盘,1,299,已完成
1008,2026-07-04,王五,显示器,1,1299,已取消
1009,2026-07-05,李四,键盘,2,299,已完成
1010,2026-07-05,赵六,耳机,1,399,已完成

保存为 orders.csv。

第三步:写分析脚本

#!/usr/bin/env python3
"""用 DuckDB 分析 CSV 数据"""
import duckdb
import sys
from pathlib import Path

def analyze_csv(csv_path):
    """读取 CSV 并输出分析结果"""
    con = duckdb.connect()

    # 自动识别 CSV 结构
    result = con.execute(f"""
        SELECT * FROM read_csv_auto('{csv_path}')
    """).fetchdf()

    # 1. 基础统计
    print(f"\n📊 文件: {csv_path}")
    print(f"   总行数: {len(result)}")
    print(f"   列数: {len(result.columns)}")
    print(f"   列名: {list(result.columns)}")

    # 2. 按状态统计
    try:
        status_stats = con.execute(f"""
            SELECT status, COUNT(*) as count,
                   SUM(price * quantity) as total_amount
            FROM read_csv_auto('{csv_path}')
            GROUP BY status
            ORDER BY total_amount DESC
        """).fetchdf()
        print("\n📈 按状态统计:")
        for _, row in status_stats.iterrows():
            print(f"   {row['status']}: {row['count']} 单, 金额 ¥{row['total_amount']:,.0f}")
    except Exception:
        pass

    # 3. 按日期统计
    try:
        date_stats = con.execute(f"""
            SELECT date, COUNT(*) as orders,
                   SUM(price * quantity) as revenue
            FROM read_csv_auto('{csv_path}')
            WHERE status = '已完成'
            GROUP BY date
            ORDER BY date
        """).fetchdf()
        print("\n📅 每日销售(已完成):")
        for _, row in date_stats.iterrows():
            print(f"   {row['date']}: {row['orders']} 单, ¥{row['revenue']:,.0f}")
    except Exception:
        pass

    # 4. TOP 产品
    try:
        top_products = con.execute(f"""
            SELECT product, COUNT(*) as times,
                   SUM(quantity) as total_qty,
                   SUM(price * quantity) as revenue
            FROM read_csv_auto('{csv_path}')
            WHERE status = '已完成'
            GROUP BY product
            ORDER BY revenue DESC
            LIMIT 5
        """).fetchdf()
        print("\n🏆 TOP 产品:")
        for i, (_, row) in enumerate(top_products.iterrows(), 1):
            print(f"   #{i} {row['product']}: {row['total_qty']} 件, ¥{row['revenue']:,.0f}")
    except Exception:
        pass

    con.close()

if __name__ == '__main__':
    csv_file = sys.argv[1] if len(sys.argv) > 1 else 'orders.csv'
    analyze_csv(csv_file)

代码不长,核心就三步:

  1. read_csv_auto() 让 DuckDB 自动识别 CSV 列类型
  2. 用 SQL 做分组、聚合、排序
  3. 把结果打印出来

第四步:运行分析

python3 analyze_csv.py orders.csv

输出类似这样:

📊 文件: orders.csv
   总行数: 10
   列数: 7
   列名: ['order_id', 'date', 'customer', 'product', 'quantity', 'price', 'status']

📈 按状态统计:
   已完成: 8 单, 金额 ¥5,562
   已取消: 2 单, 金额 ¥1,598

📅 每日销售(已完成):
   2026-07-01: 2 单, ¥977
   2026-07-02: 1 单, ¥1,299
   2026-07-03: 2 单, ¥576
   2026-07-04: 1 单, ¥299
   2026-07-05: 2 单, ¥996

🏆 TOP 产品:
   #1 键盘: 4 件, ¥1,196
   #2 显示器: 1 件, ¥1,299
   #3 鼠标: 4 件, ¥356
   #4 耳机: 2 件, ¥798

第五步:扩展成通用工具

把脚本改造成支持任意 CSV 文件。你只需要传一个文件路径,它自动识别列名,做基础分析。

如果想输出到文件:

python3 analyze_csv.py orders.csv > report.txt

或者加个参数支持指定输出格式:

import json

# 把分析结果存为 JSON
output = {
    'file': csv_path,
    'rows': len(result),
    'columns': list(result.columns),
    'status_stats': status_stats.to_dict('records'),
    'date_stats': date_stats.to_dict('records'),
}
with open('report.json', 'w') as f:
    json.dump(output, f, ensure_ascii=False, indent=2)

运行效果

跑完之后你有一个通用的 CSV 分析工具。不管是销售数据、用户注册记录、还是服务器日志,丢进去都能快速出报表。

实际场景举例:

  • 财务报销:把每月的发票 CSV 丢进去,按项目/类别汇总
  • 电商数据:分析订单、客单价、复购率
  • 用户数据:统计注册量、活跃度、留存率
  • 服务器日志:分析访问量、错误分布、Top URL

不用打开 Excel 翻来翻去,一行命令出结果。

优化方向

已经能用了,想继续加也可以:

  1. Web 界面:用 Streamlit 搭个上传页面,拖拽 CSV 就能分析
  2. 自定义 SQL:让用户自己写 SQL 查询,不只是预设的统计
  3. 图表输出:用 matplotlib 生成柱状图、折线图,直接存成 PNG
  4. 多文件合并:支持一次分析多个 CSV,自动合并汇总
  5. 定时任务:每天自动分析最新的 CSV 文件,推送报告到微信/邮箱
  6. SQL 模板库:存几个常用分析模板,一键套用

最核心的就是一行 read_csv_auto(),DuckDB 帮你做了类型推断,不需要手动定义表结构。

5 分钟搭一个能用的数据分析工具,比打开 Excel 做透视表快多了。试试?

这篇文章对你有帮助吗?