Postgres跨库迁移遇CSV未终止引号字段错误求助

场景说明 #

用Python、pandas、psycopg2处理10-20百万行的Postgres跨库表迁移：分块读取源库数据→每行拼接内容计算MD5并新增列→通过copy_expert调用COPY语句（分隔符\x1f）加载到目标库。目前遇到两个问题：

• 文本字段中的未闭合双引号触发错误：psycopg2.errors.BadCopyFileFormat: unterminated CSV quoted field
• 行拼接时指定的\x1f分隔符未生效

不想删除引号，求可行解决方案。

具体解决方法 #

1. 禁用COPY的引号解析（核心解决报错问题） #

Postgres的COPY默认把双引号当作字段引用标识，碰到未闭合的就会报错。直接在COPY语句里指定QUOTE '\0'（用空字符作为引号符，等同于关闭引号解析），这样Postgres会直接读取字段原始内容，不再处理双引号。

修改后的COPY语句示例：

COPY target_table (col1, col2, ..., md5_col) FROM STDIN WITH (FORMAT CSV, DELIMITER E'\x1f', QUOTE '\0')

2. 确保分隔符生效（解决拼接问题） #

检查你的get_data_iterator和StringIteratorIO实现，避免pandas默认的CSV处理逻辑篡改分隔符或添加引号。建议手动拼接每行数据，示例代码：

import hashlib
from io import StringIO
import pandas as pd

def get_formatted_data(chunk):
    # 计算MD5列
    chunk['md5_col'] = chunk.apply(
        lambda row: hashlib.md5(''.join(str(col) for col in row).encode('utf-8')).hexdigest(),
        axis=1
    )
    # 手动用\x1f分隔所有列，生成每行字符串
    lines = []
    for _, row in chunk.iterrows():
        line = '\x1f'.join(str(val) for val in row.values) + '\n'
        lines.append(line)
    # 返回StringIO对象给copy_expert
    return StringIO(''.join(lines))

# 分块处理并加载
for chunk in pd.read_sql_query("SELECT * FROM source_table", source_conn, chunksize=10000):
    data_io = get_formatted_data(chunk)
    target_cursor.copy_expert(
        "COPY target_table FROM STDIN WITH (FORMAT CSV, DELIMITER E'\x1f', QUOTE '\0')",
        data_io
    )
target_conn.commit()

3. 备选方案：用execute_batch批量插入 #

如果COPY的格式问题始终无法解决，可改用psycopg2的execute_batch做批量插入，虽然速度略逊于COPY，但能彻底规避CSV格式的坑：

from psycopg2.extras import execute_batch
import pandas as pd
import hashlib

for chunk in pd.read_sql_query("SELECT * FROM source_table", source_conn, chunksize=10000):
    # 计算MD5列
    chunk['md5_col'] = chunk.apply(
        lambda row: hashlib.md5(''.join(str(col) for col in row).encode('utf-8')).hexdigest(),
        axis=1
    )
    # 转换为元组列表
    records = chunk.to_records(index=False).tolist()
    # 批量插入
    execute_batch(
        target_cursor,
        "INSERT INTO target_table VALUES (%s, %s, ..., %s)",  # 对应所有字段
        records
    )
    target_conn.commit()

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