使用PostgresHook.bulk_load导入CSV至PostgreSQL时遇psycopg2格式错误

问题根源 #

你的错误核心在于PostgresHook.bulk_load方法默认生成的COPY命令缺少必要参数：

• Postgres的COPY命令默认使用制表符作为字段分隔符，但你的CSV是逗号分隔，导致整行被识别为第一个字段的值，后续字段无数据，触发missing data for column 'impressions'错误。
• 默认不会跳过CSV的表头行，表头字符串会被尝试插入数值型字段，进一步加剧格式不匹配问题。

解决方案 #

放弃使用bulk_load，改用copy_expert方法自定义COPY命令，明确指定CSV格式、分隔符和表头处理逻辑，同时修正文件指针位置问题：

修改后的导入代码 #

import os
from airflow.providers.postgres.hooks.postgres import PostgresHook

# 假设get_dates()和path已定义
with open(os.path.join(path, f"{get_dates()}.csv"), 'w+', encoding='utf-8') as f:
    f.write(resultcsv)
    f.seek(0)  # 将文件指针移到文件开头，确保COPY能读取到内容
    try:
        hook = PostgresHook(postgres_conn_id='test_db')
        # 自定义COPY命令，匹配CSV格式和表结构
        hook.copy_expert(
            sql="""
                COPY test_nikita2 (campaign, impressions, clicks, cost)
                FROM STDIN
                WITH (FORMAT CSV, HEADER, DELIMITER ',', ENCODING 'UTF8')
            """,
            file=f
        )
    finally:
        pass

关键参数说明 #

• FORMAT CSV：告诉Postgres处理的是CSV格式文件
• HEADER：跳过第一行表头
• DELIMITER ','：指定逗号为字段分隔符
• 显式指定字段列表(campaign, impressions, clicks, cost)：确保CSV列和表字段顺序完全匹配（Postgres创建表时未加双引号的字段会自动转为小写，与CSV表头一致）

额外说明 #

你之前尝试修改分隔符和字段类型无效，是因为这些修改没有传递到Postgres的COPY命令中——bulk_load方法不会读取你手动设置的分隔符，始终使用默认的制表符分隔。

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