📖 本篇简介：本篇把 pandas 真正接到生产数据库上，覆盖 SQLAlchemy 连接、建表 DDL、read_sql 读取、to_sql 写入、参数化查询防 SQL 注入、chunksize 大批量写入、pandas 操作与 SQL 语句对照表，以及时区/类型映射/字符集/事务四大常见坑。前置知识：已掌握第 22 章的数据库读写基础、会写基础 SELECT/JOIN/GROUP BY。学完后你能独立搭一条"CSV → 清洗 → 入库 → SQL 反查 → 报表"的数据管道，并自信应对面试中"pandas 怎么连 MySQL"这类题。
前面章节的数据源都是 CSV/Excel。真实生产环境里，90% 的数据躺在关系型数据库里。本篇以 MySQL 为例，带你走通 pandas 与数据库的完整读写链路。若本机没有 MySQL，用 SQLite 也能跑通全部示例——SQLAlchemy 会把方言差异屏蔽掉，只需换一行连接串。
目录
1. 环境准备与连接串
2. 建库建表：DDL 示例
3. read_sql：从数据库读 DataFrame
4. to_sql：把 DataFrame 写回数据库
5. 参数化查询：防 SQL 注入
6. pandas 操作 ↔ SQL 语句对照表
7. 大批量写入：chunksize 优化
8. 四大常见坑：时区 / 类型 / 字符集 / 事务
9. 完整实战：CSV → 清洗 → 入库 → 反查
10. 本章小结
1. 环境准备与连接串
1.1 安装依赖
# 必装：SQLAlchemy（ORM 引擎，pandas 底层用它连数据库）
pip install sqlalchemy
# MySQL 官方驱动（二选一）
pip install pymysql # 纯 Python，安装简单，推荐入门
# pip install mysqlclient # C 扩展，性能更好，但 Windows 编译麻烦
# 如果本机没有 MySQL，用 SQLite 演示（Python 自带，无需安装）
说明：pandas 自己不直接连数据库，它通过 SQLAlchemy Engine 统一对接各种数据库。你只要换一行连接串，代码就能从 SQLite 切到 MySQL/PostgreSQL。
1.2 连接串格式
from sqlalchemy import create_engine
# ===== MySQL 连接串 =====
# 格式：mysql+pymysql://用户名:密码@主机:端口/数据库名?字符集
engine_mysql = create_engine(
"mysql+pymysql://root:123456@127.0.0.1:3306/testdb?charset=utf8mb4",
echo=False, # True 会打印所有 SQL，调试用
pool_size=5, # 连接池大小，生产建议 5~10
pool_recycle=3600, # 连接回收秒数，避满 MySQL 的 wait_timeout
)
# ===== SQLite 连接串（本机无 MySQL 时用这个演示）=====
# 格式：sqlite:///文件路径（三个斜杠 + 相对/绝对路径）
engine_sqlite = create_engine("sqlite:///demo.db")
# ===== 后续所有示例用 engine 变量统一指代 =====
# MySQL 场景下把下面这行换成 engine_mysql 即可，业务代码一行不用改
engine = engine_sqlite
关键：本篇所有代码都用 engine 这个变量。MySQL 场景下替换连接串即可，read_sql / to_sql 的写法完全一样。1.3 测试连接
# 执行一条原始 SQL，验证连接通不通
with engine.connect() as conn:
result = conn.exec_driver_sql("SELECT 1 AS ok")
print(result.fetchone()) # (1,)
2. 建库建表：DDL 示例
2.1 用 pandas 直接建表（最简单）
很多人不知道：`to_sql` 第一次写入时会自动建表，不需要手写 DDL。但生产环境建议显式建表，控制字段类型。
import pandas as pd
# 先造一份样例数据
users = pd.DataFrame({
"user_id": [1, 2, 3, 4],
"name": ["张三", "李四", "王五", "赵六"],
"city": ["上海", "北京", "上海", "深圳"],
"age": [28, 35, 22, 40],
})
# to_sql 第一次写：自动建表（if_exists="replace" 表示每次重建）
users.to_sql("users", engine, if_exists="replace", index=False)
print("users 表已创建，行数：", len(users))
2.2 手写 DDL（生产推荐）
from sqlalchemy import text
# 用 SQLAlchemy 的 text() 包原始 SQL，防止注入
DDL_USERS = """
CREATE TABLE IF NOT EXISTS users (
user_id INT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
city VARCHAR(50),
age INT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
DDL_ORDERS = """
CREATE TABLE IF NOT EXISTS orders (
order_id INT PRIMARY KEY AUTO_INCREMENT,
user_id INT,
amount DECIMAL(10,2),
order_date DATE,
FOREIGN KEY (user_id) REFERENCES users(user_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
# SQLite 不支持 ENGINE/FOREIGN KEY 语法，下面这句只在 MySQL 执行
# engine.connect().execute(text(DDL_USERS))
# engine.connect().execute(text(DDL_ORDERS))
# SQLite 兼容写法（本篇演示用）
with engine.begin() as conn: # begin() 会自动提交事务
conn.execute(text("""
CREATE TABLE IF NOT EXISTS users (
user_id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
city TEXT,
age INTEGER
)
"""))
conn.execute(text("""
CREATE TABLE IF NOT EXISTS orders (
order_id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
amount REAL,
order_date DATE
)
"""))
print("两张表已创建（SQLite 语法；MySQL 换成上面 DDL_USERS/DDL_ORDERS 即可）")
MySQL vs SQLite DDL 差异：
- 类型：INTvsINTEGER、VARCHARvsTEXT、DATETIMEvsTIMESTAMP；
- 引擎：MySQL 要 ENGINE=InnoDB DEFAULT CHARSET=utf8mb4，SQLite 没有；- 自增：MySQL 用AUTO_INCREMENT，SQLite 用AUTOINCREMENT。
3. read_sql：从数据库读 DataFrame
3.1 全表读取
# 方式 1：直接传表名（最简单，但会 SELECT *）
df1 = pd.read_sql("users", engine)
print(df1)
# 方式 2：写 SQL（推荐，只取需要的列）
df2 = pd.read_sql("SELECT user_id, name, city FROM users", engine)
print(df2)
3.2 带参数的查询
# 参数化查询：用 :name 占位，params 传字典（防注入，详见第 5 节）
df = pd.read_sql(
"SELECT * FROM users WHERE city = :c AND age > :min_age",
engine,
params={"c": "上海", "min_age": 25},
)
print(df)
3.3 分块读取（大表）
# chunksize=10000：每次只读 1 万行，返回一个迭代器
# 适合表很大（千万行）、内存装不下的场景
chunk_iter = pd.read_sql("SELECT * FROM orders", engine, chunksize=10000)
total = 0
for chunk in chunk_iter: # chunk 本身就是一个小 DataFrame
total += chunk["amount"].sum()
print("orders 表总金额：", total)
3.4 只读前 N 行看结构
# 调试时先看 5 行，别 SELECT * 拉全表
preview = pd.read_sql("SELECT * FROM users LIMIT 5", engine)
print(preview.dtypes)
4. to_sql：把 DataFrame 写回数据库
4.1 if_exists 三种模式
import pandas as pd
new_users = pd.DataFrame({
"user_id": [5, 6],
"name": ["钱七", "孙八"],
"city": ["杭州", "成都"],
"age": [30, 28],
})
# if_exists 三种模式：
# "fail" 表已存在就报错（默认）
# "replace" 先 DROP 再 CREATE（清空重写）
# "append" 追加到已有表（最常用）
new_users.to_sql("users", engine, if_exists="append", index=False)
# 验证
print(pd.read_sql("SELECT * FROM users", engine))
4.2 index 参数
# index=False：不要把 DataFrame 的行号写进数据库（绝大多数情况都要 False）
# index=True：会多一列叫 "index"，通常是你不想要的
new_users.to_sql("users", engine, if_exists="append", index=False)
4.3 dtype 指定字段类型
# 写入时显式指定列类型，避免 pandas 自动推断出错（比如把 VARCHAR 推断成 TEXT）
from sqlalchemy import Integer, String, Float
new_users.to_sql(
"users", engine, if_exists="append", index=False,
dtype={
"user_id": Integer(),
"name": String(50),
"city": String(50),
"age": Integer(),
},
)
5. 参数化查询：防 SQL 注入
5.1 错误示范：拼接 SQL（高危）
# ❌ 绝对不要这样写！
def get_user_bad(city: str):
# 直接把用户输入拼进 SQL 字符串——经典 SQL 注入漏洞
sql = f"SELECT * FROM users WHERE city = '{city}'"
return pd.read_sql(sql, engine)
# 如果用户输入是 "上海' OR '1'='1"，拼出来的 SQL 就变成：
# SELECT * FROM users WHERE city = '上海' OR '1'='1'
# 结果：把全表数据都吐出来了！
5.2 正确姿势：参数化查询
# ✅ 正确：用占位符 + params 字典
def get_user_good(city: str, min_age: int = 0):
# :city 是命名占位符，pandas/SQLAlchemy 会自动转义
sql = "SELECT * FROM users WHERE city = :city AND age >= :min_age"
return pd.read_sql(sql, engine, params={"city": city, "min_age": min_age})
print(get_user_good("上海", 25))
# 即使 city = "上海' OR '1'='1"，也只会查 city 恰好等于这个字符串的行，不会注入
5.3 不同数据库的占位符
| 数据库 | 占位符风格 | 示例 |
|--------|------------|------|
| SQLAlchemy 统一写法 | 命名占位 | :city |
| SQLite | ? 或 :name | WHERE city = ? |
| MySQL（pymysql） | %s 或 %(name)s | WHERE city = %s |
| psycopg2 (PG) | %s 或 %(name)s | WHERE city = %s |
建议：一律用 SQLAlchemy 的:name命名占位符 +params字典。这样换数据库不用改 SQL。
6. pandas 操作 ↔ SQL 语句对照表
| 业务需求 | SQL | pandas |
|----------|-----|--------|
| 全表查询 | SELECT * FROM users | pd.read_sql("users", engine) |
| 选列 | SELECT name, age FROM users | pd.read_sql("SELECT name, age FROM users", engine) |
| WHERE 过滤 | SELECT * FROM users WHERE age > 30 | df[df.age > 30] |
| 排序 | SELECT * FROM users ORDER BY age DESC | df.sort_values("age", ascending=False) |
| LIMIT | SELECT * FROM users LIMIT 10 | df.head(10) |
| DISTINCT | SELECT DISTINCT city FROM users | df["city"].unique() |
| GROUP BY | SELECT city, AVG(age) FROM users GROUP BY city | df.groupby("city")["age"].mean() |
| HAVING | SELECT city, COUNT(*) c FROM users GROUP BY city HAVING c > 1 | g = df.groupby("city").size(); g[g > 1] |
| INNER JOIN | SELECT * FROM orders o JOIN users u ON o.user_id=u.user_id | pd.merge(orders, users, on="user_id", how="inner") |
| LEFT JOIN | SELECT * FROM orders o LEFT JOIN users u ON ... | pd.merge(orders, users, on="user_id", how="left") |
| 聚合 SUM | SELECT SUM(amount) FROM orders | df["amount"].sum() |
| 聚合 COUNT | SELECT COUNT(*) FROM orders | len(df) 或 df.shape[0] |
| 子查询 | SELECT * FROM users WHERE user_id IN (SELECT user_id FROM orders) | df[df.user_id.isin(orders.user_id)] |
| 窗口函数 | ROW_NUMBER() OVER (PARTITION BY city ORDER BY age DESC) | df.groupby("city")["age"].rank(ascending=False) |
# 举个完整例子：用 pandas 等价于下面这条 SQL
# SELECT city, COUNT(*) AS cnt, AVG(age) AS avg_age
# FROM users
# WHERE age >= 18
# GROUP BY city
# HAVING cnt >= 1
# ORDER BY cnt DESC
result = (
users[users.age >= 18]
.groupby("city", as_index=False)
.agg(cnt=("user_id", "count"), avg_age=("age", "mean"))
.query("cnt >= 1")
.sort_values("cnt", ascending=False)
)
print(result)
7. 大批量写入：chunksize 优化
7.1 为什么要分块
一次性 to_sql 写 100 万行，会把所有数据打包成一个巨型 INSERT 语句，数据库瞬间被打爆。`chunksize` 让 pandas 分批提交。
import numpy as np
# 造 50 万行模拟数据
big = pd.DataFrame({
"user_id": range(1, 500_001),
"name": [f"user_{i}" for i in range(1, 500_001)],
"city": np.random.choice(["上海","北京","广州","深圳"], 500_000),
"amount": np.random.uniform(10, 1000, 500_000).round(2),
})
# ❌ 错误：一次性写 50 万行，可能 OOM 或锁表
# big.to_sql("big_table", engine, if_exists="replace", index=False)
# ✅ 正确：每批 5000 行提交一次
big.to_sql(
"big_table", engine,
if_exists="replace", # 第一次先建表
index=False,
chunksize=5000, # 关键：每批 5000 行
method="multi", # 用多值 INSERT，比逐行快 10 倍
)
print("50 万行已分块写入")
7.2 chunksize 怎么选
| 数据量 | chunksize 建议 |
|--------|----------------|
| < 1 万行 | 不分块，一次写完 |
| 1 万 ~ 100 万 | 5000 ~ 10000 |
| > 100 万 | 10000 ~ 50000，配合 method="multi" |
经验：chunksize 不是越大越好。太大 = 单条 SQL 太长、锁表久；太小 = 网络往返多。5000~10000 是甜点区。
7.3 method 参数
- 默认：逐行 INSERT（最慢）；
method="multi"：多值 INSERT（一条 SQL 插多行，快 5~10 倍）；- MySQL 还可以用
method="multi"配合executemany。
8. 四大常见坑：时区 / 类型 / 字符集 / 事务
8.1 时区问题
# 坑：MySQL 存的是 UTC，pandas 读出来变成 naive datetime，差 8 小时
# 解决 1：连接串里指定时区
# mysql+pymysql://root:pw@host:3306/db?charset=utf8mb4&local_infile=1
# 解决 2：读出来后手动本地化
df["order_date"] = pd.to_datetime(df["order_date"]).dt.tz_localize("Asia/Shanghai")
8.2 类型映射
| pandas 类型 | MySQL 推荐类型 | 注意 |
|-------------|----------------|------|
| int64 | BIGINT / INT | 小心 NULL：int 列有 NULL 会变 float64 |
| float64 | DOUBLE / DECIMAL | 金额用 DECIMAL(10,2)，别用 FLOAT |
| bool | TINYINT(1) | MySQL 没有原生 BOOL |
| datetime64 | DATETIME / TIMESTAMP | 时区问题见 8.1 |
| string/object | VARCHAR / TEXT | 中文务必用 utf8mb4 |
| category | VARCHAR + 索引 | 高基数字符串别用 category 入库 |
# 坑：pandas 的 int 列一旦有 NaN，自动变 float64（int 不支持 NaN）
# 写库时会变成 DOUBLE，而不是 INT
# 解决：写库前 fillna 或转成 Int64（pandas 的可空整数类型）
df["age"] = df["age"].astype("Int64") # 注意大写 I
8.3 字符集
# 坑：MySQL 默认 latin1 存中文乱码
# 解决 1：建库时指定
# CREATE DATABASE testdb DEFAULT CHARSET utf8mb4;
# 解决 2：连接串加 charset
# "mysql+pymysql://root:pw@host:3306/db?charset=utf8mb4"
# 解决 3：表级别指定
# CREATE TABLE users (...) CHARSET=utf8mb4;
8.4 事务提交
# 坑：to_sql 默认会自动提交，但用 engine.begin() 手动控制时容易忘 commit
from sqlalchemy import text
# ✅ 正确：with engine.begin() 会自动 commit / rollback
with engine.begin() as conn:
conn.execute(text("DELETE FROM users WHERE user_id = 999"))
# 出异常会自动 rollback，正常结束自动 commit
# ❌ 危险：engine.connect() 不会自动提交，需要手动 commit
# conn = engine.connect()
# conn.execute(text("DELETE FROM users WHERE user_id = 999"))
# conn.commit() # 忘了这行，数据根本没删！
9. 完整实战：CSV → 清洗 → 入库 → 反查
import pandas as pd
from sqlalchemy import create_engine, text
# 1. 造一份 CSV（实际项目里是 pd.read_csv("orders.csv")）
orders_csv = pd.DataFrame({
"order_id": [1, 2, 3, 4, 5, 6],
"user_id": [1, 1, 2, 2, 3, 3],
"amount": [100, 200, 150, np.nan, 300, 250],
"city": ["上海", "上海", "北京", "北京", "上海", "北京"],
"date": ["2025-01-05","2025-01-06","2025-01-07","2025-01-08","2025-01-09","2025-01-10"],
})
# 2. 清洗
orders_csv["amount"] = pd.to_numeric(orders_csv["amount"], errors="coerce")
orders_csv = orders_csv.dropna(subset=["amount"]) # 删掉金额缺失的行
orders_csv["date"] = pd.to_datetime(orders_csv["date"])
# 3. 入库
engine = create_engine("sqlite:///ecommerce.db")
orders_csv.to_sql("orders", engine, if_exists="replace", index=False, chunksize=1000)
# 4. 用 SQL 反查：每个城市的订单总额
sql_result = pd.read_sql("""
SELECT city,
COUNT(*) AS order_cnt,
SUM(amount) AS total_amount
FROM orders
GROUP BY city
ORDER BY total_amount DESC
""", engine)
print("=== SQL 反查结果 ===")
print(sql_result)
# 5. 等价的 pandas 写法（对比）
pd_result = (
orders_csv.groupby("city", as_index=False)
.agg(order_cnt=("order_id", "count"), total_amount=("amount", "sum"))
.sort_values("total_amount", ascending=False)
)
print("=== pandas 等价结果 ===")
print(pd_result)
=== SQL 反查结果 ===
city order_cnt total_amount
0 上海 3 600.0
1 北京 2 400.0
=== pandas 等价结果 ===
city order_cnt total_amount
1 上海 3 600.0
0 北京 2 400.0
体会：同一份数据，SQL 和 pandas 都能算。区别在于——
- 数据已经在数据库里 → 用 SQL 聚合完再 read_sql 拉结果，省内存；- 数据已经在 pandas 里 → 直接 pandas 操作，省一次 IO。
10. 本章小结
10.1 速查
| 操作 | API |
|------|-----|
| 连接 | create_engine("mysql+pymysql://user:pw@host:3306/db?charset=utf8mb4") |
| 读全表 | pd.read_sql("table", engine) |
| 读 SQL | pd.read_sql("SELECT ...", engine, params={...}) |
| 写表 | df.to_sql("table", engine, if_exists="append", index=False) |
| 分块读 | pd.read_sql(..., chunksize=10000) |
| 分块写 | df.to_sql(..., chunksize=5000, method="multi") |
| 防注入 | 永远用 params 传参，别拼 SQL |
10.2 第三方库清单
pandas：DataFrame 主体；sqlalchemy：连接引擎（必装）；pymysql：MySQL 驱动（纯 Python，Windows 友好）；mysqlclient：MySQL C 驱动（可选，更快但难装）；- SQLite：Python 自带，无需安装。
10.3 练习
1. 把本篇的 SQLite 示例换成 MySQL 连接串，在本机 MySQL 上跑通。
2. 造 100 万行数据，分别用默认 to_sql 和 chunksize=5000, method="multi" 写入，对比耗时。
3. 用参数化查询实现一个函数 get_orders_by_city(city, start_date, end_date)，打印 SQL 语句观察占位符。
4. 故意拼一条 f"... WHERE city='{city}'" 的 SQL，传入 city="上海' OR '1'='1"，观察结果差异。
5. 把第 32 章题 10 的 pandas 版本改写成先 read_sql 入库、再用 SQL 聚合的版本。
下一篇预告：第 34 章 pandas 与 PostgreSQL 实战：大结果集与连接池 —— 进阶到 PG，掌握流式查询、连接池配置与大结果集不爆内存。
文章回复
0 条公开回复