使用psycopg2+pandas批量导入CSV至PostgreSQL遇ValueError问题求助

问题根源 #

你遇到的QueryCanceledError本质是因为混合使用了文件的迭代操作（next(f)）和读取操作（copy_from内部调用的read()）。next()是通过文件的迭代器接口读取数据，而copy_from会直接调用文件对象的read()方法，这两种操作会干扰文件指针的位置，导致Python无法正确追踪剩余数据，最终抛出数据丢失的错误。

直接解决方案：替换next(f)为f.readline() #

最简单的修复是把跳过表头的方式从next(f)改为f.readline()——后者属于文件的读取方法，和copy_from内部的操作兼容，不会触发指针冲突：

with open(pathCSV, 'r') as f:
    # 读取并丢弃表头行
    f.readline()
    # 直接执行数据导入
    cur.copy_from(f, tableName, sep=',')

如果你的psycopg2版本在2.7及以上，还可以更简洁地使用skip_header参数（无需手动处理文件指针）：

with open(pathCSV, 'r') as f:
    cur.copy_from(f, tableName, sep=',', skip_header=1)

优化后的完整脚本 #

除了修复文件读取的问题，我还对你的脚本做了一些健壮性优化（比如数据类型映射、错误处理、表名安全处理等）：

import pandas as pd
import psycopg2
import os

# 定义pandas数据类型到PostgreSQL的映射函数
def map_dtype(pandas_dtype):
    if pd.api.types.is_integer_dtype(pandas_dtype):
        return 'integer'
    elif pd.api.types.is_float_dtype(pandas_dtype):
        return 'numeric'
    elif pd.api.types.is_string_dtype(pandas_dtype):
        return 'varchar(255)'
    elif pd.api.types.is_datetime64_dtype(pandas_dtype):
        return 'timestamp'
    else:
        return 'text'

try:
    # 建立数据库连接
    conn = psycopg2.connect("host=localhost dbname=somedb user=postgres password=somepw")
    cur = conn.cursor()

    # 目标CSV目录
    csv_dir = r"C:\Data\Waste_Intervention\Census_Tables\Cleaned"
    # 仅筛选CSV文件，避免处理其他格式
    csv_files = [f for f in os.listdir(csv_dir) if f.lower().endswith('.csv')]

    for file_name in csv_files:
        file_path = os.path.join(csv_dir, file_name)
        # 生成安全的表名：去掉后缀、替换空格为下划线、转为小写
        table_name = os.path.splitext(file_name)[0].replace(' ', '_').lower()
        # 如果你坚持原表名逻辑，可以用：table_name = file_name.split("_")[-1][:-4]

        # 读取CSV获取结构
        df = pd.read_csv(file_path)

        # 构建CREATE TABLE的列定义
        column_definitions = []
        for idx, (col_name, dtype) in enumerate(df.dtypes.items()):
            pg_type = map_dtype(dtype)
            # 将第一列设为主键
            if idx == 0:
                pg_type += ' PRIMARY KEY'
            # 用双引号包裹列名，避免和SQL关键字冲突
            column_definitions.append(f'"{col_name}" {pg_type}')

        # 创建表（IF NOT EXISTS避免重复创建报错）
        create_stmt = f'CREATE TABLE IF NOT EXISTS {table_name} ({", ".join(column_definitions)})'
        cur.execute(create_stmt)
        print(f"处理表：{table_name} - 已创建/验证结构")

        # 导入数据
        with open(file_path, 'r', encoding='utf-8') as f:
            f.readline()  # 跳过表头
            # null参数指定CSV中空值的表示（比如空字符串）
            cur.copy_from(f, table_name, sep=',', null='')
        
        print(f"处理表：{table_name} - 数据导入完成")

    # 提交所有变更
    conn.commit()
    print("所有CSV文件处理完成！")

except Exception as e:
    # 出错时回滚事务
    if conn:
        conn.rollback()
    print(f"执行出错：{str(e)}")
finally:
    # 确保关闭游标和连接
    if cur:
        cur.close()
    if conn:
        conn.close()

关键优化点说明 #

1. 数据类型映射：不再硬编码为integer和numeric，而是根据pandas的实际数据类型自动匹配PostgreSQL类型，适配更多场景
2. 安全表名：处理文件名中的空格和特殊字符，避免SQL语法错误
3. 错误处理：添加try-except-finally块，确保出错时回滚事务，并正确释放数据库资源
4. 兼容空值：通过null参数指定CSV中空值的处理方式，避免导入时的空值错误

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