第 34 章 pandas 与 PostgreSQL 实战：大结果集与连接池

📖 本篇简介：本篇讲解 pandas 连接 PostgreSQL（psycopg2 / SQLAlchemy）的工程化实战，覆盖类型映射、大结果集分块与流式读取、SQLAlchemy 连接池配置、COPY 批量导入导出，以及与 MySQL 的差异对照。前置知识：第 4 章基础读写、第 22 章数据库读写优化、第 33 章。学完后你能让 pandas 在生产库上既不撑爆内存、也不打满连接数，安全高效地完成"查询 → 清洗 → 回写"全链路。

第 33 章讲了 pandas 连 MySQL。PostgreSQL（简称 PG）在类型系统、流式查询和批量导入上与 MySQL 差异不小：PG 有原生 array/jsonb/hstore 类型、服务端游标、COPY 命令，工程上更适合"大查询 + 高并发连接"。本篇按 连接 → 类型映射 → 大结果集 → 连接池 → 批量导入 → 差异对比 → 常见坑 的顺序展开。

目录

1. 连接 PostgreSQL：psycopg2 与 SQLAlchemy

2. PG 类型与 pandas dtype 映射表

3. 大结果集：分块读取与流式游标

4. 连接池：QueuePool 配置与复用

5. COPY 批量导入导出

6. 与 MySQL 的差异对比

7. 常见坑与注意事项

8. 本章小结与练习

1. 连接 PostgreSQL：psycopg2 与 SQLAlchemy

1.1 安装驱动

# 基础三件套：SQLAlchemy 负责连接池与方言，psycopg2 是 PG 官方推荐驱动
pip install "sqlalchemy>=2.0" psycopg2-binary pandas

# psycopg2-binary 是预编译版，免装 PG 开发头文件；
# 如果需要从源码编译（极少场景），改装 psycopg2 并确保本机有 pg_config。

本机无 PG 怎么办：本篇所有"连接池 / 分块 / 回写"逻辑都通过 SQLAlchemy 编写，把连接串换成 sqlite:///demo.db 即可在本机跑通等价流程。文中凡涉及 PG 专属语法（COPY、RETURNING、::jsonb）都会明确标注。

1.2 用 SQLAlchemy 建立引擎（推荐）

from sqlalchemy import create_engine, text
import pandas as pd

# 连接串格式：postgresql+驱动://用户名:密码@主机:端口/库名
engine = create_engine(
    "postgresql+psycopg2://analytics:********@pg.internal:5432/orders_db",
    echo=False,            # True 时打印所有 SQL，调试用；生产务必关掉
    pool_pre_ping=True,    # 取连接前先 ping 一下，避免拿到"已被服务端杀掉"的死连接
    future=True,           # 使用 SQLAlchemy 2.0 风格 API
)

# 试连一句（SQLAlchemy 2.x 里原生 SQL 必须包 text()）
with engine.connect() as conn:
    ver = conn.execute(text("SELECT version()")).scalar()
print(ver)
# PostgreSQL 16.1 on x86_64-pc-linux-gnu ...

1.3 读：read_sql 一把梭

# 1) 整表小数据：直接 read_sql_table（自动带 schema）
df_user = pd.read_sql_table("users", engine, schema="public",
                            columns=["id", "city", "created_at"])

# 2) 自定义 SQL + 参数化查询（防 SQL 注入，永远不要拼字符串）
sql = text("""
    SELECT user_id, amount, pay_time
    FROM orders
    WHERE city = :city AND pay_time >= :since
""")
df_order = pd.read_sql(sql, engine, params={"city": "上海", "since": "2025-01-01"})

# 3) 写：to_sql 回写，if_exists 三选一
df_order.to_sql("orders_clean", engine, schema="mart",
                if_exists="append",   # fail / replace / append
                index=False,
                chunksize=5000)      # 大批量写入时按 5000 行一批提交，避免单条 INSERT 太慢

安全：pd.read_sql 的 params 用命名占位符 :name（SQLAlchemy 方言），psycopg2 原生游标则用 %(name)s。两种占位符不要混用。

2. PG 类型与 pandas dtype 映射表

PG 的类型系统比 MySQL 丰富，读入 pandas 后 dtype 常与直觉不符，先看映射表：

| PostgreSQL 类型 | pandas dtype | 说明 / 注意点 |

|-----------------|--------------|---------------|

| smallint / int2 | int16 | 范围 -32768~32767，够用时优先用，省内存 |

| integer / int4 | int32 | 默认整型 |

| bigint / int8 | int64 | 主键自增到超大表时通常是它 |

| real / float4 | float32 | 精度低，金额列别用 |

| double precision / float8 | float64 | 默认浮点 |

| numeric / decimal(p,s) | object | 任意精度小数会变成 Python Decimal 对象列，计算前要 astype(float) |

| boolean | bool | NULL 会变 boolean（可空布尔），pandas 1.0+ |

| varchar / text | object（或 string） | 建议 dtype_backend="pyarrow" 时读成 string |

| date | datetime64[ns] | 时间部分为 0 点 |

| timestamp（无时区） | datetime64[ns] | 直接对齐 |

| timestamptz（带时区） | datetime64[ns, UTC] | 跨时区排坑重点，见 7.3 |

| json / jsonb | object | 读进来是 Python dict/list，需 json_normalize 展开 |

| uuid | object | 读进来是 uuid.UUID 对象，可转 str |

| integer[]（数组） | object | 读进来是 Python list |

| bytea | object | 二进制，pandas 一般不直接分析 |

import pandas as pd

# 常见清洗：numeric 列转 float、timestamptz 转本地时区、jsonb 拍平
df = pd.read_sql("SELECT * FROM orders LIMIT 1000", engine)

# 1) numeric(12,2) 列读成 object，先转 float 才能做均值/求和
df["amount"] = df["amount"].astype(float)

# 2) timestamptz 读成带 tz 的 datetime，转 Asia/Shanghai 做后续日切分
df["pay_time"] = df["pay_time"].dt.tz_convert("Asia/Shanghai")

# 3) jsonb 列读成 dict，用 json_normalize 拍平成多列
meta = pd.json_normalize(df["extra"].tolist())
df = pd.concat([df.drop(columns=["extra"]), meta.add_prefix("extra_")], axis=1)

print(df.dtypes)
# amount                      float64
# pay_time    datetime64[ns, Asia/Shanghai]
# extra_channel               object
# ...

坑：numeric 列若不转 float，分组聚合时会极慢（Python 对象级循环），且 pd.to_numeric 比 astype(float) 更宽容（能顺带修脏数据）。

3. 大结果集：分块读取与流式游标

生产库一张表几千万行是常态。pd.read_sql("SELECT * FROM big", engine) 不带任何限制，会把整张表拉进内存，直接 OOM。三种正确姿势：

3.1 read_sql + chunksize（最简单）

# chunksize=10_000 时，read_sql 返回一个迭代器，每次吐 1 万个行组成的小 DataFrame
# 内存占用恒定 ≈ 单个 chunk 的大小，与结果集总行数无关
agg_total = 0.0
agg_cnt = 0

for chunk in pd.read_sql(
    "SELECT amount FROM orders WHERE pay_time >= '2025-01-01'",
    engine,
    chunksize=10_000,          # 每块行数，按内存大小调，通常 5_000~50_000
):
    agg_total += chunk["amount"].sum()
    agg_cnt += len(chunk)

print(f"总行数={agg_cnt}, 金额合计={agg_total:,.2f}")

chunksize 适合"逐块聚合 / 逐块过滤后落盘"。不要把所有 chunk 攒到一个 list 再 concat——那就退化成全量加载了。

3.2 服务端游标：psycopg2 named cursor（百万行级推荐）

chunksize 本质是"客户端拉一行、攒一批"，PG 服务端默认会把结果一次性发到客户端，网络内存压力都在客户端。named cursor 让 PG 在服务端分批吐数据：

import psycopg2
import pandas as pd

# 直连 psycopg2（不走 SQLAlchemy 连接池时的写法；工程上建议把它包进连接池里）
conn = psycopg2.connect("host=pg.internal dbname=orders_db user=analytics")

# name 一给，psycopg2 自动变成"服务端游标"；itersize 控制每批从 PG 拉多少行
with conn.cursor(name="order_stream") as cur:
    cur.itersize = 5_000       # 服务端每批吐 5000 行到客户端缓冲
    cur.execute("SELECT user_id, amount, pay_time FROM orders ORDER BY id")

    # 转成 DataFrame 逐块消费
    while True:
        rows = cur.fetchmany(5_000)    # 与 itersize 对齐；也可以 cur 直接迭代
        if not rows:
            break
        chunk = pd.DataFrame(rows, columns=["user_id", "amount", "pay_time"])
        # ...对 chunk 做聚合 / 过滤 / 写盘...
        print("本块行数", len(chunk))

conn.close()

3.3 SQLAlchemy 流式读取（stream_results）

# SQLAlchemy 2.x：在连接层面开 streaming，效果等同服务端游标
from sqlalchemy import text

with engine.connect().execution_options(stream_results=True) as conn:
    result = conn.execution_options(stream_results=True).execute(
        text("SELECT user_id, amount FROM orders WHERE pay_time >= :since"),
        {"since": "2025-01-01"},
    )
    # 逐行迭代，内存里永远只有一行（+ 驱动内部小缓冲）
    for row in result.partitions(5_000):      # 每批 5000 行
        chunk = pd.DataFrame(row, columns=result.keys())
        # ...逐块处理...
        _ = chunk

三种方式怎么选：

| 方式 | 适用 | 内存 | 写法 |

|------|------|------|------|

| read_sql(chunksize=) | 中小结果集、快速上手 | 客户端攒批 | 最简单 |

| psycopg2 named cursor | 千万行级、纯 PG | 服务端攒批 | 略繁琐 |

| SQLAlchemy stream_results | 已用 SQLAlchemy 生态 | 服务端攒批 | 与 ORM 一致 |

4. 连接池：QueuePool 配置与复用

pandas 每次 pd.read_sql(sql, engine) 都"新开连接"是灾难——TCP 握手 + PG 鉴权 + 后端进程 fork，单次几十毫秒。SQLAlchemy 默认用 QueuePool 维护一个连接池，连接用完归还而非关闭。

4.1 关键参数

engine = create_engine(
    "postgresql+psycopg2://analytics:********@pg.internal:5432/orders_db",
    # ---- 池大小 ----
    pool_size=10,             # 常驻连接数，常用值 5~20，按 PG 的 max_connections 留余量
    max_overflow=10,          # 突发时允许临时多开的连接数，峰值 = pool_size + max_overflow = 20
    pool_timeout=30,          # 池满时排队等连接的秒数，超时抛 TimeoutError
    # ---- 保活 ----
    pool_recycle=1800,        # 连接最大复用秒数，超过就销毁重建，防止被防火墙/PG wait_timeout 掐断
    pool_pre_ping=True,       # 取连接前 SELECT 1 探活，死连接自动丢弃（强烈建议开）
)

| 参数 | 作用 | 推荐值 |

|------|------|--------|

| pool_size | 空闲常驻连接数 | 5~20 |

| max_overflow | 允许临时超出 pool_size 的连接 | 0~10 |

| pool_recycle | 连接寿命（秒） | 1800（30 分钟） |

| pool_pre_ping | 取连接前探活 | True |

| pool_timeout | 等连接超时（秒） | 30 |

4.2 正确复用引擎，而不是重复创建

# ❌ 错误：每个函数都 create_engine，连接池形同虚设，连接数瞬间爆炸
def query_orders_bad():
    eng = create_engine("postgresql+psycopg2://...")
    return pd.read_sql("SELECT * FROM orders LIMIT 10", eng)

# ✅ 正确：引擎做成模块级单例，整个进程共享同一个池
engine = create_engine("postgresql+psycopg2://...", pool_size=10, max_overflow=10)

def query_orders():
    return pd.read_sql("SELECT * FROM orders LIMIT 10", engine)

# 进程退出前显式回收（脚本最后一行）
# engine.dispose()

4.3 常见异常：连接池耗尽

sqlalchemy.exc.TimeoutError: QueuePool limit of size 10 overflow 10 reached,
connection timed out: 30

含义：池里 20 条连接全被占用且 30 秒内没归还。排查方向：

1. 是不是某段代码拿了连接没关（with engine.connect() 漏写）？

2. 是不是长事务没提交（to_sql 在某些旧版 pandas 里不显式 commit）？

3. 是不是并发任务（多线程 / multiprocessing）把池占满了？多进程下每个进程各自有一份池，总连接数 = 进程数 × 20，要在 PG max_connections 内算好。

5. COPY 批量导入导出

COPY 是 PG 最快的批量装卸方式，比 INSERT 快一个数量级（少了 SQL 解析、WAL 逐条提交）。psycopg2 用 copy_expert 在 pandas 里直接驱动。

5.1 从 DataFrame 高速入库

import io
import psycopg2

conn = psycopg2.connect("host=pg.internal dbname=orders_db user=analytics")
cur = conn.cursor()

def df_to_pg_copy(df, table, conn):
    # 1) 先把 DataFrame 序列化成 PG COPY 认识的 CSV 文本（不走磁盘，走内存 StringIO）
    buf = io.StringIO()
    df.to_csv(buf, index=False, header=False)   # COPY 模式不需要表头
    buf.seek(0)

    # 2) COPY FROM：告诉 PG 从"stdin"读 CSV，列顺序与 df 列对齐
    sql = f"""
        COPY {table} ({', '.join(df.columns)})
        FROM STDIN WITH (FORMAT csv)
    """
    cur.copy_expert(sql, buf)
    conn.commit()                                # COPY 也在事务里，必须显式提交

# 准备一个百万行 DataFrame（这里用模拟数据演示）
import pandas as pd, numpy as np
big = pd.DataFrame({
    "user_id": np.arange(1_000_000),
    "amount": np.random.uniform(1, 1000, 1_000_000).round(2),
})

# 实测：百万行 COPY 通常 1~3 秒，逐条 to_sql 要 30 秒以上
df_to_pg_copy(big, "orders_clean", conn)
cur.close(); conn.close()

5.2 从 PG 高速导出到 DataFrame

conn = psycopg2.connect("host=pg.internal dbname=orders_db user=analytics")
cur = conn.cursor()

buf = io.StringIO()
# COPY TO STDOUT：把查询结果直接吐到客户端缓冲区
cur.copy_expert("COPY (SELECT user_id, amount FROM orders WHERE amount > 500) TO STDOUT WITH (FORMAT csv, HEADER true)", buf)
buf.seek(0)

# 再把 CSV 文本读回 pandas——比 read_sql 快得多，尤其列多的时候
df = pd.read_csv(buf)
print(df.head())

局限：COPY FROM STDIN / TO STDOUT 需要文件级权限且走的是服务端文件协议的 stdin 变体；远程只读账号通常允许 COPY ... TO STDOUT（导出），而 COPY table TO '/server/path' 是服务端文件权限，远程用户一般没资格。

6. 与 MySQL 的差异对比

| 维度 | PostgreSQL | MySQL | pandas 侧注意点 |

|------|-----------|-------|------------------|

| 官方驱动 | psycopg2 / psycopg3 | PyMySQL / mysqlclient | 连接串前缀不同：postgresql+psycopg2:// vs mysql+pymysql:// |

| 占位符 | %(name)s（原生）/ :name（SQLAlchemy） | %s / %(name)s | 统一走 SQLAlchemy :name 最省心 |

| 标识符引号 | 双引号 "users" | 反引号 users | 列名带空格时写法不同 |

| 分页 | LIMIT 50 OFFSET 100 | LIMIT 100, 50 | 用 SQLAlchemy limit/offset 方法生成方言无关 SQL |

| 自增主键 | BIGSERIAL / GENERATED ALWAYS AS IDENTITY | AUTO_INCREMENT | to_sql 建表时类型自动转 |

| 时间类型 | timestamptz 原生带时区 | 一般存 datetime 无时区 | 跨时区数据 PG 更友好 |

| 半结构化 | jsonb（可索引、可展开） | json（功能弱） | PG 读 jsonb 直接 dict，MySQL json 列常要 JSON_EXTRACT |

| 批量导入 | COPY（极快） | LOAD DATA INFILE | pandas 都要走驱动 API |

| 大小写敏感 | 字符串默认敏感（取决于 collation，多数 PG 默认敏感） | 默认不敏感 | 去重 / join 前注意 |

| 服务端游标 | 原生支持 named cursor | 支持但需 SSCursor | 大结果集都建议流式 |

# 同一套 pandas 代码，换连接串就能换库——这就是 SQLAlchemy 方言的价值
PG_URL = "postgresql+psycopg2://analytics:********@pg.internal:5432/orders_db"
MYSQL_URL = "mysql+pymysql://report:********@mysql.internal:3306/orders_db"

eng_pg = create_engine(PG_URL, pool_pre_ping=True)
eng_my = create_engine(MYSQL_URL, pool_pre_ping=True)

# SQL 用标准方言写（LIMIT/OFFSET），pandas 自动翻译成各自方言
df_pg = pd.read_sql("SELECT city, SUM(amount) s FROM orders GROUP BY city ORDER BY s DESC LIMIT 10", eng_pg)
df_my = pd.read_sql("SELECT city, SUM(amount) s FROM orders GROUP BY city ORDER BY s DESC LIMIT 10", eng_my)

7. 常见坑与注意事项

7.1 驱动安装失败

Error: pg_config executable not found.     # 装 psycopg2（编译版）时报错

解决：生产 / 学习环境一律用 psycopg2-binary，它自带预编译二进制，不需要 pg_config。

7.2 事务没提交，连接被占着

# ❌ 旧版 pandas to_sql 在某些驱动下不会自动 commit，
#    导致连接一直处于"未提交事务"状态，连接池里这条永远不可用。
df.to_sql("tmp", engine)

# ✅ 显式包一层事务（SQLAlchemy 2.x 写法）
with engine.begin() as conn:               # with 块结束自动 commit，异常自动 rollback
    df.to_sql("tmp", conn, if_exists="replace", index=False)

7.3 timestamptz 时区漂移

# PG 里存的是 timestamptz = '2025-01-01 10:00+08'，读进来是 UTC 时间
df["pay_time"] = pd.read_sql("SELECT pay_time FROM orders", engine)["pay_time"]
print(df["pay_time"].head())
# 0   2025-01-01 02:00:00+00:00     ← 被转成 UTC 了！

# 做"按本地日期切分"之前一定要 tz_convert 回业务时区
df["pay_time_local"] = df["pay_time"].dt.tz_convert("Asia/Shanghai")
df["pay_date"] = df["pay_time_local"].dt.date

7.4 流式读取忘了 ORDER BY / 大事务

• 服务端游标在一个事务里持有，事务越长 PG 的元组锁 / snapshot 越重，读完立刻 `commit` 释放；
• cur.execute("SELECT ...") 后忘 commit，下一次查询会拿到旧快照。

7.5 内存控制速查

| 操作 | 危险写法 | 安全写法 |

|------|----------|----------|

| 读大表 | pd.read_sql("SELECT * FROM big", eng) | chunksize= 或 stream_results=True |

| 读大表全列 | SELECT * | 只 SELECT 需要的列 |

| 写大表 | df.to_sql("t", eng) 一把梭 | chunksize=5000 或 COPY |

| 多进程并发 | 每进程 pool_size=20 | 算好总量 ≤ PG max_connections |

8. 本章小结与练习

小结

• 连接：统一用 SQLAlchemy.create_engine，连接串 postgresql+psycopg2://；本机无 PG 用 SQLite 跑通逻辑；
• 类型：numeric → object、timestamptz → 带时区 datetime、jsonb → dict，三处最易踩坑；
• 大结果集：chunksize（简单）、named cursor / stream_results（千万行级），永远不要 SELECT * 全量拉；
• 连接池：pool_size + max_overflow + pool_recycle + pool_pre_ping，引擎做模块级单例；
• 批量：百万行以上用 copy_expert 的 COPY FROM/TO STDIN；
• 与 MySQL：占位符、引号、分页、自增语法不同，靠 SQLAlchemy 方言屏蔽。

练习题

1. 写一个 get_engine(db_url) 工厂函数，返回配好 pool_pre_ping / pool_recycle 的引擎，并在进程退出时 dispose。

2. 用 chunksize=5000 分块读一张千万行表，只统计某列的 sum/count/max，全程内存占用恒定。

3. 把一份 50 万行 DataFrame 用 copy_expert 入库，对比 to_sql(chunksize=5000) 的耗时。

4. 给一张 timestamptz 列的表写代码：读入后转 Asia/Shanghai，按天聚合 GMV。

5. 模拟连接池耗尽：在多线程里开 30 个并发查询，观察 TimeoutError，再调大 pool_size/max_overflow 解决。

下一篇预告：第 35 章 特征工程与数据预处理 —— 缺失值、编码、缩放、时间特征、文本特征与防数据泄漏的完整 sklearn 链路。