优化从两个PostgreSQL大表批量获取数据的方案咨询

你的代码效率低的核心问题是每次执行LIMIT查询时没有维护读取位置，PostgreSQL会每次从头扫描全表，随着已读取数据量增加，数据库需要遍历越来越多的无效数据，导致耗时指数上升，和批量大小、表切换关系不大。下面是几个可行的优化方案：

方案1：使用服务器端游标（推荐） #

psycopg2的服务器端游标会在数据库端保存读取位置，避免重复全表扫描，每次fetch都从当前位置获取下一批数据，效率极高。

import psycopg2
from psycopg2 import extras

# 为两个表分别创建独立连接，避免游标位置干扰
conn_a = psycopg2.connect(dbname="xxx", user="xxx", password="xxx", host="xxx", port=5432)
conn_b = psycopg2.connect(dbname="xxx", user="xxx", password="xxx", host="xxx", port=5432)

# 创建服务器端游标
cur_a = conn_a.cursor(name='cursor_a', cursor_factory=extras.NamedTupleCursor)
cur_b = conn_b.cursor(name='cursor_b', cursor_factory=extras.NamedTupleCursor)

batch_size = 100000

# 执行全表查询，由服务器端游标维护位置
cur_a.execute("SELECT * FROM table_a")
cur_b.execute("SELECT * FROM table_b")

while True:
    # 交替获取批量数据
    rows_a = cur_a.fetchmany(batch_size)
    rows_b = cur_b.fetchmany(batch_size)

    if not rows_a and not rows_b:
        break

    # 处理数据逻辑
    # process_rows(rows_a)
    # process_rows(rows_b)

cur_a.close()
cur_b.close()
conn_a.close()
conn_b.close()

注意：服务器端游标建议为每个表单独使用数据库连接，避免不同游标的位置互相干扰。

方案2：基于有序唯一键的分页查询 #

如果表有自增主键（比如id）或者其他带索引的有序唯一列，可以用WHERE id > last_id LIMIT batch_size的方式，利用索引快速定位下一批数据，彻底避免全表扫描。

import psycopg2

conn = psycopg2.connect(dbname="xxx", user="xxx", password="xxx", host="xxx", port=5432)
cur = conn.cursor()

batch_size = 100000
last_id_a = 0
last_id_b = 0

while True:
    # 从table_a获取下一批数据
    cur.execute("SELECT * FROM table_a WHERE id > %s ORDER BY id LIMIT %s", (last_id_a, batch_size))
    rows_a = cur.fetchall()
    if rows_a:
        last_id_a = rows_a[-1][0]  # 假设id是结果集的第一列

    # 从table_b获取下一批数据
    cur.execute("SELECT * FROM table_b WHERE id > %s ORDER BY id LIMIT %s", (last_id_b, batch_size))
    rows_b = cur.fetchall()
    if rows_b:
        last_id_b = rows_b[-1][0]

    if not rows_a and not rows_b:
        break

    # 处理数据逻辑
    # process_rows(rows_a)
    # process_rows(rows_b)

cur.close()
conn.close()

要求：表必须有带索引的有序唯一列，否则WHERE id > %s会退化为全表扫描，无法提升效率。

方案3：并行异步获取数据 #

用Python的asyncio配合asyncpg（异步PostgreSQL驱动），同时从两个表异步读取数据，减少网络等待时间，提升整体吞吐量。

import asyncio
import asyncpg

async def fetch_batch(conn, table_name, batch_size):
    batch = []
    async for record in conn.cursor(f"SELECT * FROM {table_name}"):
        batch.append(record)
        if len(batch) == batch_size:
            yield batch
            batch = []
    if batch:
        yield batch

async def main():
    conn_a = await asyncpg.connect(dbname="xxx", user="xxx", password="xxx", host="xxx", port=5432)
    conn_b = await asyncpg.connect(dbname="xxx", user="xxx", password="xxx", host="xxx", port=5432)

    batch_size = 100000
    task_a = asyncio.create_task(fetch_batch(conn_a, "table_a", batch_size))
    task_b = asyncio.create_task(fetch_batch(conn_b, "table_b", batch_size))

    active_tasks = [task_a, task_b]
    while active_tasks:
        done, pending = await asyncio.wait(active_tasks, return_when=asyncio.FIRST_COMPLETED)
        for task in done:
            try:
                batch = await task
                # 处理数据逻辑
                # process_rows(batch)
            except StopAsyncIteration:
                active_tasks.remove(task)

    await conn_a.close()
    await conn_b.close()

asyncio.run(main())

优势：异步IO可以同时处理两个表的读取请求，有效减少网络等待时间，适合数据库与应用服务器跨网络部署的场景。

内容的提问来源于stack exchange，提问作者user23121263