Pandas 读取 SQL 进阶

Pandas 读取 SQL 进阶：事务、原生SQL执行、多表关联、游标、批量写入优化

前置依赖：承接上一篇《Pandas 读取 SQL 数据库》，依然使用 pandas + sqlalchemy，数据库示例以 MySQL 为主，大部分写法兼容 PostgreSQL / SQLite。 重点解决普通 read_sql_query 覆盖不到的场景：DDL、DML、事务控制、复杂多表、游标流式、海量数据写入调优、冲突更新（upsert）。

pip install pandas pymysql sqlalchemy

import pandas as pd
from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError

连接引擎复用：

engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test_db?charset=utf8mb4")

一、区分：read_sql 只适合查询；execute 执行 DDL/DML

pd.read_sql_query 只能执行 SELECT，返回 DataFrame。 如果需要建表、删表、更新、删除、执行存储过程，要用 connection.execute(text(sql))。

必须导入 text，把字符串SQL包装成SQLAlchemy文本语句，新版本推荐写法。

with engine.connect() as conn:
    # 建临时表示例
    conn.execute(text("""
    CREATE TABLE IF NOT EXISTS sale_tmp (
        id INT PRIMARY KEY AUTO_INCREMENT,
        order_no VARCHAR(32),
        goods_name VARCHAR(64),
        qty INT,
        price DECIMAL(10,2)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
    """))
    # 手动提交（autocommit默认关闭）
    conn.commit()

坑：在 with connect() 块内，DML不会自动提交，忘记 commit() 会看上去执行成功，但数据库无变化。

二、事务控制（重点：多SQL原子执行，要么全部成功，要么回滚）

业务场景：先查询、再删除旧数据、再写入新汇总数据；中间报错不能出现一半数据。 使用 .begin() 开启事务。

sql_del = text("DELETE FROM sale_summary WHERE stat_date = :stat_date")
sql_insert = text("""
INSERT INTO sale_summary(stat_date, goods_name, total_qty, total_amount)
VALUES(:stat_date, :goods_name, :total_qty, :total_amount)
""")

stat_date = "2026-10-01"
df_sum = pd.DataFrame([
    {"stat_date": stat_date, "goods_name":"笔记本", "total_qty":10, "total_amount":350},
    {"stat_date": stat_date, "goods_name":"钢笔", "total_qty":20, "total_amount":160}
])

try:
    with engine.begin() as conn:
        # begin() 上下文：自动事务，无异常自动commit，异常自动rollback
        conn.execute(sql_del, {"stat_date": stat_date})
        # 批量插入多条
        conn.execute(sql_insert, df_sum.to_dict("records"))
        print("事务执行成功")
except SQLAlchemyError as e:
    print("事务失败，自动回滚：", e)

✅ engine.begin() 和 engine.connect() 区别：

• connect()：手动 commit / rollback
• begin()：托管事务，推荐业务数据同步场景

三、多表关联查询写法

方式1：数据库JOIN（推荐，性能最好，数据库索引生效）

SQL内完成多表join，pandas只接收最终结果，适合大数据。

sql_join = text("""
SELECT d.order_no, d.goods_name, d.qty, d.price, c.customer_name, c.phone
FROM sale_detail d
LEFT JOIN customer c ON d.customer_id = c.id
WHERE d.sale_date >= :start_date
""")
df_join = pd.read_sql_query(sql_join, engine, params={"start_date":"2026-01-01"})

方式2：Pandas内存合并（小表场景，不推荐大表）

两张表分别读成DataFrame，用 merge 在内存关联；数据量大时极易OOM、无法走索引。

df_detail = pd.read_sql_query("SELECT * FROM sale_detail limit 2000", engine)
df_cust = pd.read_sql_query("SELECT id, customer_name, phone FROM customer", engine)
df_merge = pd.merge(
    df_detail,
    df_cust,
    left_on="customer_id",
    right_on="id",
    how="left"
)

选型口诀：能在SQL JOIN，就不要拿到Pandas里merge。

四、原生游标流式读取（比chunksize更底层，可控性更强）

pd.read_sql_query(chunksize=xxx) 底层也是游标封装；如果需要自定义游标、自定义fetch逻辑，可以直接拿到cursor。 适合：需要逐行处理、自定义日志、实时进度打印、过滤丢弃部分行。

from sqlalchemy import text

sql = text("SELECT id, order_no, goods_name, qty, price FROM sale_detail")

with engine.connect() as conn:
    result = conn.execute(sql)
    rows = []
    for row in result:
        # row 是命名元组，可以 row.goods_name 取值
        rows.append(dict(row))
        if len(rows) >= 1000:
            df_chunk = pd.DataFrame(rows)
            print(f"处理批次，行数 {len(df_chunk)}")
            rows.clear()
    # 处理剩余不足一批的数据
    if rows:
        df_chunk = pd.DataFrame(rows)
        print(f"最后批次，行数 {len(df_chunk)}")

注意：SQLAlchemy Result 对象是惰性加载，不会一次性把全表载入内存。

五、to_sql 批量写入深度优化（解决慢、锁表、性能问题）

基础写法 df.to_sql 在数据上万、十万行时会很慢，下面是全套优化手段。

5.1 核心参数说明回顾

df.to_sql(
    name="sale_backup",
    con=engine,
    if_exists="append",
    index=False,
    chunksize=2000,
    method="multi"
)

• chunksize：每批次提交行数
• method="multi"：一次execute提交多行，减少网络往返（MySQL支持）不写 method="multi" 默认单行循环插入，速度差很多！

⚠️ 限制：multi 不适合PostgreSQL高版本，PG推荐默认+调大chunksize。

5.2 高性能模式：原生executemany（比to_sql更快，可控）

海量数据场景，绕开pandas to_sql封装，直接用 executemany：

df_write = pd.DataFrame([...])
insert_sql = text("""
INSERT INTO sale_backup(order_no, goods_name, qty, price)
VALUES (:order_no, :goods_name, :qty, :price)
""")
data_list = df_write.to_dict("records")

with engine.begin() as conn:
    # executemany批量提交
    conn.execute(insert_sql, data_list)

5.3 UPSERT：存在则更新，不存在则插入（MySQL ON DUPLICATE KEY UPDATE）

业务经常需要：主键唯一，重复时更新，而不是报错。to_sql 原生不支持upsert，需要自己写SQL。 前提：表有唯一索引/主键。

upsert_sql = text("""
INSERT INTO sale_summary(stat_date, goods_name, total_qty, total_amount)
VALUES (:stat_date, :goods_name, :total_qty, :total_amount)
ON DUPLICATE KEY UPDATE
    total_qty = VALUES(total_qty),
    total_amount = VALUES(total_amount)
""")

data = [
    {"stat_date":"2026-10-01", "goods_name":"笔记本", "total_qty":12, "total_amount":420}
]
with engine.begin() as conn:
    conn.execute(upsert_sql, data)

PostgreSQL 对应语法 ON CONFLICT ... DO UPDATE，语法不一样。

5.4 写入避坑清单

1. 关闭自动提交循环、优先 engine.begin() 事务包裹一批数据；
2. 大表写入避开业务高峰，防止锁表；
3. 目标表提前建好索引，不要让to_sql自动建表（自动建表字段类型经常不合理）；
4. 超百万行：不要to_sql，优先导出csv + LOAD DATA INFILE（MySQL最快方案）。

六、存储过程调用

with engine.connect() as conn:
    # 调用存储过程 proc_sale_stat，传入参数
    result = conn.execute(text("CALL proc_sale_stat(:date_arg)"), {"date_arg":"2026-10-01"})
    df_proc = pd.DataFrame(result.fetchall())
    print(df_proc)

七、超时、连接池配置（长时间脚本、定时任务必配）

默认连接池参数很小，定时脚本循环查询很容易报 queuepool timeout。

engine = create_engine(
    "mysql+pymysql://root:123456@127.0.0.1:3306/test_db?charset=utf8mb4",
    pool_size=10,          # 池中常驻连接数
    max_overflow=20,       # 额外临时连接上限
    pool_recycle=3600,    # 1小时回收连接，断开mysql超时长连接
    pool_pre_ping=True    # 取用连接前ping一下，剔除失效连接
)

定时脚本、长期运行的ETL程序，必须加 pool_recycle + pool_pre_ping。

八、完整ETL实战：数据库取数→清洗→汇总→upsert回写

承接前面销售报表业务链路

import pandas as pd
from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError

engine = create_engine(
    "mysql+pymysql://root:123456@127.0.0.1:3306/test_db?charset=utf8mb4",
    pool_size=5, max_overflow=10, pool_recycle=3600, pool_pre_ping=True
)

start_date = "2026-10-01"
end_date = "2026-10-01"

sql_read = text("""
SELECT order_no, goods_name, qty, price, sale_date
FROM sale_detail
WHERE sale_date BETWEEN :s AND :e
""")

try:
    with engine.connect() as conn:
        df = pd.read_sql_query(sql_read, con=conn, params={"s": start_date, "e": end_date}, parse_dates=["sale_date"])
    print(f"读取原始明细 {len(df)} 行")

    # 清洗、汇总
    df["amount"] = df["qty"] * df["price"]
    df["stat_date"] = df["sale_date"].dt.date.astype(str)
    df_summary = df.groupby(["stat_date", "goods_name"]).agg(
        total_qty=("qty", "sum"),
        total_amount=("amount", "sum")
    ).reset_index()

    # upsert写入汇总表
    upsert_sql = text("""
    INSERT INTO sale_summary(stat_date, goods_name, total_qty, total_amount)
    VALUES (:stat_date, :goods_name, :total_qty, :total_amount)
    ON DUPLICATE KEY UPDATE
        total_qty = VALUES(total_qty),
        total_amount = VALUES(total_amount)
    """)
    data_rows = df_summary.to_dict("records")
    with engine.begin() as conn:
        conn.execute(upsert_sql, data_rows)
    print("汇总数据回写完成")

except SQLAlchemyError as err:
    print("ETL任务异常：", err)

九、进阶API速查表

场景	推荐写法
普通查询返回df	pd.read_sql_query(text(sql), con, params={})
DDL / UPDATE / DELETE	conn.execute(text(sql), params) + commit
原子批量任务	with engine.begin() as conn:
流式逐行处理	conn.execute(sql) + for row in result
快速批量插入	to_sql(method="multi", chunksize=2000)
存在更新、不存在插入	手写 ON DUPLICATE KEY / ON CONFLICT
防止连接失效	pool_recycle + pool_pre_ping

十、进阶避坑汇总

1. 不要混用 read_sql 和裸execute在同一个事务里，容易连接状态错乱；
2. method="multi" 只是MySQL优化，PG/Oracle不要强行使用；
3. 事务块内尽量少放逻辑，数据库事务不宜长时间持有；
4. 百万级数据不要用pandas作为中间层，优先数据库原生导出导入；
5. 日期参数传参直接传字符串/date对象，不要自己拼接格式化；
6. 定时任务一定要配置连接池回收，不然长时间运行连接耗尽。

小结

本篇进阶覆盖了工业级ETL最常用的能力：事务保证数据一致性、多表关联选型、游标流式读取、高性能写入和Upsert。 常规分析用 read_sql_query；定时同步、数据回流、汇总回写、修改表结构，就用这套SQLAlchemy事务+execute方案。