Pandas读写SQL数据库教程：read_sql与to_sql用法详解

本节目标：学会用 pandas 连接 SQL 数据库，执行查询、把 DataFrame 写回数据库，掌握参数化查询、写入策略和连接管理。

pandas 与数据库：一座桥

企业数据大多住在数据库里：MySQL、PostgreSQL、SQLite……pandas 提供一组函数充当桥梁，让查询结果直接变成 DataFrame，也让 DataFrame 直接写进数据库表。

核心就四个函数：

函数	作用
`pd.read_sql()`	通用入口：传 SQL 就执行查询，传表名就读整表
`pd.read_sql_query()`	只接受 SQL 语句，执行查询
`pd.read_sql_table()`	只接受表名，读取整张表
`df.to_sql()`	把 DataFrame 写入数据库表

实际工作中最常用 read_sql，它会根据传入内容自动判断。

pandas 自己不直接连数据库，连接对象由第三方库创建后传进来。主流的两种：SQLAlchemy 引擎（推荐）和 sqlite3 原生连接（轻量）。

两种连接方式

SQLAlchemy 是 Python 最流行的数据库工具包，用一个连接字符串搞定各种数据库：

from sqlalchemy import create_engine

# SQLite：文件数据库，无需用户名密码
engine = create_engine("sqlite:///mydata.db")

# MySQL / PostgreSQL（需要装对应驱动）
# engine = create_engine("mysql+pymysql://root:password@localhost:3306/mydb")
# engine = create_engine("postgresql+psycopg2://user:password@localhost:5432/mydb")

连接字符串的格式：数据库+驱动://用户名:密码@主机:端口/库名。连 MySQL 需要 pip install pymysql，PostgreSQL 需要 psycopg2-binary。

SQLite 更简单，Python 内置 sqlite3 模块，不用装任何东西：

import sqlite3

conn = sqlite3.connect("mydata.db")      # 连文件数据库
# conn = sqlite3.connect(":memory:")     # 内存数据库，测试专用

read_sql 和 to_sql 两种连接都能用；但 read_sql_table 只支持 SQLAlchemy 引擎，用 sqlite3 连接会报 NotImplementedError。

Note

pandas 2.2 起还支持 Apache Arrow 的 ADBC 驱动（如 adbc_driver_sqlite），性能、空值处理和类型识别都更好，但要额外安装。学习阶段用 SQLAlchemy 完全够，追求极致性能再上 ADBC。

读取：read_sql

传表名，读整张表：

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///mydata.db")
df = pd.read_sql("employees", con=engine)   # 传表名

传 SQL 语句，执行查询：

df = pd.read_sql("SELECT * FROM employees WHERE department = 'IT'", con=engine)

复杂的多表查询也直接写：

sql = """
SELECT e.name, e.salary, d.department_name
FROM employees e
JOIN departments d ON e.dept_id = d.id
WHERE e.salary > 10000
ORDER BY e.salary DESC
"""
df = pd.read_sql(sql, con=engine)

常用的配套参数：

df = pd.read_sql("SELECT * FROM orders",
                 con=engine,
                 index_col="id",           # 指定列做行索引
                 parse_dates=["created_at"])  # 解析日期列

index_col 和 parse_dates 和 CSV 章节的用法一样。日期在数据库里常常是字符串，parse_dates 帮你转成 datetime64。

Warning

数据库查询结果为空时，pandas 会把所有列推断为最通用的类型，可能导致后续类型判断出错。如果空结果会影响代码，可以显式转换 dtype（§07 讲过 astype）。

参数化查询：防注入的正确姿势

查询条件来自用户输入时，千万别用字符串拼接 SQL：

# 危险写法：存在 SQL 注入风险
# dept = input("部门：")
# df = pd.read_sql(f"SELECT * FROM employees WHERE department = '{dept}'", engine)

正确做法是把参数通过 params 传进去。sqlite3 连接用 ? 占位符：

df = pd.read_sql(
    "SELECT * FROM employees WHERE department = ? AND salary > ?",
    con=engine,
    params=["IT", 8000],   # 按顺序对应两个问号
)

SQLAlchemy 用 :名称 命名占位符，配合 sqlalchemy.text()：

from sqlalchemy import text

df = pd.read_sql(
    text("SELECT * FROM employees WHERE department = :dept AND salary > :min_salary"),
    con=engine,
    params={"dept": "IT", "min_salary": 8000},
)

参数化不只是风格问题，而是安全问题。拼接字符串等于把 SQL 的执行权交给了输入者，一条 '; DROP TABLE ... 就能毁掉数据。所有外部输入必须走 params。

写入：to_sql

to_sql 把 DataFrame 写进数据库表：

df = pd.DataFrame({
    "name": ["张三", "李四", "王五"],
    "department": ["IT", "HR", "IT"],
    "salary": [12000, 8000, 15000],
})

df.to_sql("employees", con=engine, if_exists="replace", index=False)

三个关键参数：

if_exists：表已存在时的处理。"fail"（默认）报错，"replace" 删表重建，"append" 追加数据
index：默认 True，会把行索引写成一列。通常不需要，记得 index=False
chunksize：大表分批写入，比如 chunksize=1000 每次写 1000 行

if_exists="append" 是增量写入的标配：

new_employees = pd.DataFrame({
    "name": ["赵六"],
    "department": ["Finance"],
    "salary": [11000],
})
new_employees.to_sql("employees", con=engine, if_exists="append", index=False)

追加时列名和类型必须和原表匹配，否则会报错或写入失败。

写完后验证一下：

result = pd.read_sql("SELECT * FROM employees", con=engine)
print(result)

Tip

默认的 to_sql 会为每行生成一条 INSERT 语句，数据量大时偏慢。追求性能可以传 method="multi" 批量插入（部分数据库不支持），或后续用数据库自身的批量导入工具。

类型映射与读表进阶

to_sql 写库时，pandas 会把 DataFrame 的 dtype 映射成 SQL 类型。可以用 dtype 参数覆盖默认映射：

from sqlalchemy.types import String

df.to_sql("t", con=engine, if_exists="replace", index=False,
          dtype={"name": String(50)})

两个容易踩的点：

timedelta64 列会以纳秒整数写入数据库并给出警告（多数数据库没有”时间差”类型）
category 列会转成普通值写入，读回来不再是 category

read_sql_table 还能指定要读的列和 schema：

df = pd.read_sql_table("employees", con=engine,
                       columns=["id", "name", "salary"],
                       schema="hr")

schema 参数在 MySQL、PostgreSQL 这类有模式概念的数据库上有意义；SQLite 没有 schema 的概念，用不上。

大表分块读取

几百万行的表直接 SELECT * 可能撑爆内存。两个思路：

在 SQL 层过滤：WHERE、LIMIT 能筛就筛，让数据库干活
chunksize 分块：每次只拿一部分

for chunk in pd.read_sql("SELECT * FROM large_table", con=engine, chunksize=10000):
    print(chunk.shape)   # 每块 10000 行

配合 pd.concat 可以逐块处理后再合并（合并的细节在 §33 讲）。

连接管理

连接用完要关，否则可能锁库、占资源。sqlite3 连接建议用 with 自动管理：

with sqlite3.connect("mydata.db") as conn:
    df = pd.read_sql("SELECT * FROM employees", con=conn)
# 退出 with 块后连接自动关闭，df 数据仍然可用

SQLAlchemy 引擎是连接池，创建一次可以反复用，程序结束时由引擎统一清理，日常不用手动关。但每次用 engine.connect() 打开的具体连接，也要记得关或放进 with。

小结

四个函数：read_sql 通用、read_sql_query 只查、read_sql_table 只读表（仅 SQLAlchemy）、to_sql 写入
SQLAlchemy 连接字符串 数据库+驱动://用户:密码@主机:端口/库名；SQLite 用内置 sqlite3 零依赖
外部输入必须参数化：sqlite3 用 ?，SQLAlchemy 用 :name，配合 params
to_sql 记得 index=False，if_exists 控制 fail / replace / append
类型映射可用 dtype 覆盖；timedelta、category 两列有特殊行为
大表用 SQL 过滤或 chunksize 分块，连接用完要关

到这里，数据进出 pandas 的常用通道（CSV、Excel、JSON、SQL）都打通了。下一节开始进入正题：怎么高效地查看和选择数据。

SQL 数据库读写

pandas 与数据库：一座桥

两种连接方式

读取：read_sql

参数化查询：防注入的正确姿势

写入：to_sql

类型映射与读表进阶

大表分块读取

连接管理

小结