Python 数据采集与存储：CSV、JSON、SQL 实战速查

预计阅读时间：6 分钟

日常开发中，数据不会凭空出现——要么从文件读入，要么从数据库查询。CSV、JSON 和 SQL 是 Python 数据处理的三条主干道，掌握它们的读写细节，能避免大量低级错误和性能坑。

下面按格式逐一梳理关键操作，附带可直接运行的代码。

CSV：最朴素但最容易踩坑的表格格式

CSV 看起来简单，实际暗藏陷阱：编码问题、逗号出现在字段内部、换行符不一致。Python 标准库 csv 模块已经处理了大部分边缘情况，前提是你选对参数。

import csv

# 写入——用 newline='' 防止 Windows 下多出空行
with open("employees.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "dept", "salary"])
    writer.writeheader()
    writer.writerows([
        {"name": "张伟", "dept": "工程部", "salary": 28000},
        {"name": "李娜", "dept": "产品部", "salary": 32000},
    ])

# 读取——DictReader 让每行变成字典，比索引列号更安全
with open("employees.csv", "r", encoding="utf-8") as f:
    for row in csv.DictReader(f):
        print(f"{row['name']} / {row['dept']} / ¥{row['salary']}")

几个要点：

open 时务必指定 newline=''，否则 Windows 写入会多出 \r\r\n。
字段含逗号或换行时，csv 模块自动加引号处理；手动拼接字符串反而容易出错。
大文件场景考虑 pandas.read_csv，它支持分块读取（chunksize 参数）。

JSON：嵌套结构与类型映射

JSON 和 Python 的类型映射基本直觉一致——字典对应 object，列表对应 array——但有两处容易翻车：datetime 无法直接序列化，数字精度在浮点与整数间可能丢失。

import json
from datetime import datetime

data = {
    "project": "数据平台 v2",
    "created": datetime.now(),
    "members": ["张伟", "李娜", "王磊"],
    "config": {"retry": 3, "timeout": 30.5},
}

# 自定义序列化：处理 datetime
def json_encoder(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Object of type {type(obj)} is not JSON serializable")

# 写入
with open("project.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2, default=json_encoder)

# 读取
with open("project.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)
    print(loaded["created"])  # 字符串 "2025-01-15T10:30:00"，需自行转回 datetime

注意 ensure_ascii=False——不加这个参数，中文会被转成 \uXXXX，文件可读性大幅下降。

SQL：从连接到查询的完整链路

Python 操作 SQL 的标准接口是 sqlite3（内置）或第三方驱动（如 psycopg2 对 PostgreSQL、mysql.connector 对 MySQL）。核心原则：永远用参数化查询，不要拼接 SQL 字符串。

import sqlite3

# 创建数据库和表
conn = sqlite3.connect("company.db")
cur = conn.cursor()
cur.execute("""
    CREATE TABLE IF NOT EXISTS employees (
        id   INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        dept TEXT,
        salary INTEGER
    )
""")
conn.commit()

# 参数化插入——? 占位，杜绝 SQL 注入
new_rows = [("张伟", "工程部", 28000), ("李娜", "产品部", 32000)]
cur.executemany(
    "INSERT INTO employees (name, dept, salary) VALUES (?, ?, ?)",
    new_rows,
)
conn.commit()

# 查询并转为字典列表（比元组更易后续处理）
cur.execute("SELECT name, dept, salary FROM employees WHERE salary > ?", (25000,))
columns = [desc[0] for desc in cur.description]
results = [dict(zip(columns, row)) for row in cur.fetchall()]
for r in results:
    print(r)

conn.close()

切换到 PostgreSQL 或 MySQL 时，占位符从 ? 变为 %s（psycopg2）或 :1（oracle），其余逻辑不变。生产环境建议用 SQLAlchemy 或 DuckDB 统一层，减少驱动差异。

三种格式的选择逻辑

场景	推荐格式	原因
人工编辑的简单表格	CSV	Excel / 记事本都能直接打开
配置文件、API 响应、嵌套数据	JSON	结构表达力强，Python 原生映射
多条件查询、关系型关联、并发写入	SQL	索引、事务、约束是数据库的核心价值

实际项目经常混合使用：API 返回 JSON → 筛选后写入 SQL → 定期导出 CSV 给业务方。理解每种格式的边界，才能在串联时少踩坑。

实战检查清单

每次处理数据文件或数据库前，快速过一遍：

编码：open 是否指定了 encoding="utf-8"？
换行：CSV 写入是否加了 newline=''？
中文：JSON dump 是否设了 ensure_ascii=False？
安全：SQL 是否用了参数化查询，而非字符串拼接？
类型：JSON 中的日期/时间是否自定义了序列化？
关闭：数据库连接是否在操作结束后 close() 或用了 with 上下文？

六条规则覆盖了绝大多数低级错误。数据采集和存储不复杂，但细节决定成败——把上面的代码片段保存为模板，下次直接改字段名就能跑。