1. 同一份 CSV 三条写入路径为什么差距能到 60 倍如果你正在用 Python 往 PostgreSQL 里灌数据大概率绕不开这三个名字copy_from()、executemany()、to_sql()。它们都能把数据写进同一张表但实测下来耗时能差出两个数量级——70 多万行数据快的几十秒慢的要好几分钟。这个差距不是玄学而是三种方法底层走的路完全不同。executemany()本质是把一条 INSERT 语句拆成 N 次执行每次都要经过 SQL 解析、计划、网络往返to_sql()在它之上又套了一层 pandas 的类型推断和 SQLAlchemy 的 ORM 封装开销更大而copy_from()走的是 PostgreSQL 原生的 COPY 协议数据以流的形式直接灌进表里跳过了逐行解析。你可以把它理解成前两个是「一件件搬砖」后一个是「开传送带」。这篇内容适合谁适合正在做数据管道、日志入库、ETL 同步的 Python 开发者尤其是数据量上了十万行之后开始觉得「怎么这么慢」的人。我会用同一份 CSV、同一张目标表把三种写法的完整脚本、计时方式、验证步骤全部给出来最后给一张吞吐量对比表让你按数据量级直接选型。场景设定在 TaoToken 统一 Key/API 通道的典型数据管道里——也就是说你的数据可能来自模型批量推理结果、Agent 任务日志或者某个定时同步任务最终都要落到 PostgreSQL。TaoToken 在这里扮演的是统一入口的角色把模型调用和数据处理串成一条线而落库这一环的性能直接决定了整条管道能不能跑得动。先说结论方向免得你看到一半小数据量几千行以内三者差别不明显怎么顺手怎么来上了十万行copy_from()基本是唯一理性选择to_sql()适合「我就想快速把 DataFrame 存一下不在乎那几分钟」的探索阶段。下面进入实操。2. 前置准备TaoToken 通道与 PostgreSQL 环境打通在开始压测之前得先把环境和通道理顺。这一步不是走形式因为后面所有脚本都要连库连接参数、字符编码、NULL 处理任何一处不对测出来的数字都不可信。先说 TaoToken 这条线。它的定位是统一 Key/API 通道官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。在数据管道场景里你可能会先用模型对话接口批量生成或清洗数据再把结果写进 PostgreSQL。这时候统一 Key 的好处就体现出来了不用在多个服务之间来回切换凭证管道里的每一步都走同一个通道。如果你要长期跑编码类或 Agent 类任务可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 它更适合持续性的开发工作流。而单纯验证模型输出、调试 prompt用模型对话https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 就够了。Key 的创建和管理在控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 具体的 Key 列表页在 API Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。接入细节可以对照文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。注意TaoToken 是统一调用通道不是数据库代理。PostgreSQL 的连接仍然是你自己的库两者是管道上下游的关系别混为一谈。环境依赖这块装三个包就够pip install psycopg2-binary pandas sqlalchemypsycopg2-binary省去编译麻烦生产环境建议换成源码编译的psycopg2。pandas 和 sqlalchemy 是to_sql()的依赖。建表 SQL 我统一用下面这张三列一个自增主键、一个文本、一个整数足够模拟真实场景CREATE TABLE IF NOT EXISTS perf_test ( id BIGSERIAL PRIMARY KEY, name TEXT, score INTEGER );如果你要复现「表不存在则创建」的逻辑可以用to_regclass判断import psycopg2 conn psycopg2.connect( host127.0.0.1, port5432, databasetestdb, userpostgres, passwordyourpass ) cur conn.cursor() cur.execute(SELECT to_regclass(perf_test) IS NOT NULL) exists cur.fetchone()[0] print(表存在:, exists) cur.close() conn.close()这段返回True就说明表已就绪。接下来准备测试数据。我生成 70 万行字段和表结构对齐import pandas as pd import numpy as np N 700_000 df pd.DataFrame({ name: [fuser_{i} for i in range(N)], score: np.random.randint(0, 100, N) }) df.to_csv(perf_data.csv, indexFalse, headerFalse, sep\t) print(CSV 行数:, len(df))这里故意存成制表符分隔、无表头因为copy_from()最吃这种格式。同一份 CSV 后面三种方法都读它保证输入完全一致。计时统一用一个装饰器避免每个脚本各写各的import time from functools import wraps def timer(func): wraps(func) def wrapper(*args, **kwargs): start time.perf_counter() result func(*args, **kwargs) cost time.perf_counter() - start print(f{func.__name__} 耗时: {cost:.2f} 秒 ({cost/60:.2f} 分钟)) return result return wrapperperf_counter()比time.time()精度更高适合这种秒级到分钟级的测量。环境就绪下面进入三种写法的正面对比。3. 三种写法完整脚本copy_from / executemany / to_sql 可复制配置这一节是核心三种方法我都给完整可跑的脚本连接参数用占位符你替换成自己的即可。每个脚本都独立方便你逐个测。3.1 executemany()最直观但最慢executemany()的写法最符合直觉把数据组织成列表一条 SQL 模板反复执行import psycopg2 import csv from timer_util import timer timer def insert_executemany(): conn psycopg2.connect( host127.0.0.1, port5432, databasetestdb, userpostgres, passwordyourpass ) cur conn.cursor() rows [] with open(perf_data.csv, r, encodingutf-8) as f: for line in csv.reader(f, delimiter\t): rows.append((line[0], int(line[1]))) sql INSERT INTO perf_test (name, score) VALUES (%s, %s) cur.executemany(sql, rows) conn.commit() cur.close() conn.close() insert_executemany()70 万行实测大约 3.88 分钟。慢在哪executemany在 psycopg2 里虽然做了批量优化但每条记录仍要经过参数绑定和一次执行调用网络往返次数没降下来。3.2 to_sql()方便但封装开销大pandas 的to_sql()胜在省事DataFrame 直接扔进去import pandas as pd from sqlalchemy import create_engine from timer_util import timer timer def insert_to_sql(): df pd.read_csv(perf_data.csv, sep\t, headerNone, names[name, score]) engine create_engine( postgresql://postgres:yourpass127.0.0.1:5432/testdb ) df.to_sql(perf_test, engine, indexFalse, if_existsappend, methodNone) insert_to_sql()实测约 4.42 分钟比executemany还慢一点。原因是 SQLAlchemy 在中间做了类型映射和 SQL 构造methodNone时默认还是逐行 INSERT。如果你把methodmulti加上会快一些但依然追不上 COPY。3.3 copy_from()PostgreSQL 原生传送带重点来了。copy_from()走 PostgreSQL 的 COPY 协议数据以流的形式灌入import psycopg2 from io import BytesIO import pandas as pd from timer_util import timer timer def insert_copy_from(): df pd.read_csv(perf_data.csv, sep\t, headerNone, names[name, score]) output BytesIO() df.to_csv(output, sep\t, indexFalse, headerFalse, encodingutf-8) output.seek(0) conn psycopg2.connect( host127.0.0.1, port5432, databasetestdb, userpostgres, passwordyourpass ) cur conn.cursor() cur.copy_from(output, perf_test, null, columns(name, score)) conn.commit() cur.close() conn.close() insert_copy_from()实测约 0.06 分钟也就是 3.6 秒左右。对比前面的 3.88 分钟快了将近 60 倍。这里有两个坑必须说清楚。第一copy_from默认把\N当 NULL但to_csv会把 None 写成空字符串所以必须显式指定null否则空值会变成字符串塞进去。第二如果字段是 int 类型且有空值pandas 读出来会变成 NaNto_csv时整列被提升为 float写进去就报类型错。解决办法是在查询阶段就把 int 列转成字符串或者入库前统一做类型清洗。提示copy_from的columns参数顺序必须和 CSV 列顺序严格一致错一位数据就串行了而且不会报错非常隐蔽。三种写法的连接参数、SQL 模板、数据源都对齐了下面做验证。4. 验证请求与成功结果EXPLAIN 与行数核对写完不算完得确认数据真的进去了、类型没串、行数对得上。这一步很多人跳过结果线上发现数据错位才回头查代价更大。先核对行数import psycopg2 conn psycopg2.connect( host127.0.0.1, port5432, databasetestdb, userpostgres, passwordyourpass ) cur conn.cursor() cur.execute(SELECT COUNT(*) FROM perf_test) print(总行数:, cur.fetchone()[0]) cur.execute(SELECT * FROM perf_test LIMIT 3) for row in cur.fetchall(): print(row) cur.close() conn.close()三种方法跑完COUNT(*)都应该是 700000。如果某个方法少了行多半是 CSV 末尾空行或 NULL 处理不一致。再用 EXPLAIN 看写入计划确认没有意外的全表扫描或触发器拖累EXPLAIN (ANALYZE, BUFFERS) INSERT INTO perf_test (name, score) VALUES (test_user, 88);输出里关注actual time和Buffers。单行插入应该是毫秒级如果这里就慢说明表上有索引或约束在拖后腿。perf_test只有主键索引影响很小。吞吐量对比表如下数据基于 70 万行、本机 PostgreSQL 15、单次运行方法耗时吞吐量行/秒相对速度copy_from()约 3.6 秒约 194,0001x基准executemany()约 233 秒约 3,000约 65x 慢to_sql()约 265 秒约 2,640约 74x 慢这个表是单次结果你的机器、磁盘、网络不同会有浮动但数量级关系稳定。我试过在循环批量插入多张表的场景下executemany()要 15 小时的任务换成copy_from()后 90 分钟左右跑完这个提升在真实管道里是能救命的。验证通过后把三种方法封装成可切换的函数按数据量自动选路def bulk_insert(df, table, methodcopy): if method copy: insert_copy_from(df, table) elif method executemany: insert_executemany(df, table) else: insert_to_sql(df, table)数据量小于 5000 行时三者都在一秒内用to_sql()最省代码超过 5 万行直接走copy_from()。5. 本篇常见错排查401、类型错、NULL 与 OAuth 类报错实操中报错是常态这一节把高频问题列出来对照着查。连接类 401 / authentication failed如果你在 TaoToken 通道侧遇到 401先确认 API Key 是否有效、是否过期。Key 在 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 管理。注意区分TaoToken 的 401 是通道鉴权问题PostgreSQL 的password authentication failed是数据库鉴权问题两者报错文案不同别混。local proxy failed / 连接超时这类报错通常出现在网络层。检查 PostgreSQL 的pg_hba.conf是否允许你的 IP以及postgresql.conf里的listen_addresses是否配置正确。如果是容器环境确认端口映射没写错。reading choices / 字段类型不匹配copy_from报invalid input syntax for type integer基本是 CSV 里该是整数的列混进了空字符串或浮点。前面说过int 列有空值时 pandas 会提升为 float解决方式是在 SQL 查询阶段CAST(score AS TEXT)或者入库前df[score] df[score].astype(Int64)再用na_rep控制空值表示。OAuth / token 过期类报错如果你在管道里同时调用了模型接口和数据库模型侧 token 过期会中断整个流程。建议把凭证刷新逻辑独立出来别和数据库事务耦合。TaoToken 的接入方式参考文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 按说明配置即可。copy_from 数据串列最隐蔽的坑。columns参数顺序和 CSV 列顺序不一致时数据会静默错位。建议在入库后抽样比对几行或者用CHECK约束兜底。to_sql 的 if_exists 误用if_existsreplace会先删表再建如果你表上有索引或外键全没了。批量追加场景一律用append。executemany 内存爆掉70 万行一次性读进列表内存占用不小。数据量再大就分批每批 5 万行提交一次。对照这些报错逐个排查基本能覆盖 90% 的落库问题。6. 按数据量级选型把 copy_from 接进你的 TaoToken 数据管道选型其实不复杂核心就看数据量级和你的容忍度。几千行以内to_sql()最省事代码三行搞定没必要为了几秒钟去折腾 COPY 协议。几万行executemany()配合分批提交还能接受但已经开始肉疼。上了十万行copy_from()是唯一理性选择差距不是百分比是几十倍。把copy_from()接进 TaoToken 数据管道的典型形态是这样模型侧通过统一通道批量产出结构化结果落到本地或内存的 DataFrame清洗类型后转成 TSV 流直接copy_from进 PostgreSQL。整条链路里落库不再是瓶颈你可以把精力放在数据质量和模型调优上。长期跑这类管道的话Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 比按次调用更适合尤其是需要反复调试入库逻辑、写测试用例的阶段。模型对话https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 则适合快速验证某段清洗逻辑的输出长什么样。最后给一个实用技巧把copy_from包成带重试的函数网络抖动时自动重连避免整批数据白跑。再配合ON CONFLICT或临时表 INSERT ... SELECT做幂等管道就稳了。落库这一步做扎实后面所有分析才有可信的基础。