使用Apache Airflow bulk_load无法向PostgreSQL上传文件排查

1. 文件写了但读不到 #

你代码里用w+模式打开文件，写完resultcsv后，文件指针停在文件末尾，bulk_load读取时会从当前指针位置开始，自然读不到内容。

改法：写完后重置指针，或者把bulk_load移到with块外面（with结束会自动关闭文件，重新打开时指针在开头）：

path = r'/opt/airflow/dags/asd/data/'
filename = os.path.join(path, f"{get_dates()}.csv")
with open(filename, 'w+', encoding='utf-8') as f:
    f.write(resultcsv)
# 出了with块文件自动关闭，再读就是从头开始
try:
    hook = PostgresHook(postgres_conn_id='test_db')                   
    hook.bulk_load('testdb', filename)
finally:           
    os.remove(filename)

2. COPY命令参数不匹配 #

从日志看，执行的COPY命令是COPY testdb FROM STDIN，没指定分隔符、表头处理等参数。如果你的CSV文件有表头，或者用了非默认的分隔符（比如逗号以外的符号），PostgreSQL会把表头当作数据行尝试插入，类型不匹配时可能被静默忽略（取决于数据库的COPY错误处理配置）。

改法：用bulk_load_custom自定义COPY参数，比如CSV带表头、逗号分隔的场景：

hook.bulk_load_custom(
    table='testdb',
    filename=fname,
    copy_options="DELIMITER ',' CSV HEADER"
)

3. 事务未提交 #

部分情况下，PostgresHook的操作可能处于未提交的事务中，虽然日志显示任务成功，但数据未持久化到数据库。

改法：在bulk_load后显式提交事务：

hook.get_conn().commit()

4. 文件内容为空或不符合表结构 #

先确认resultcsv变量是否真的包含有效数据。可以在写入文件后添加日志打印，验证文件内容和大小：

with open(filename, 'w+', encoding='utf-8') as f:
    f.write(resultcsv)
    f.seek(0)
    print(f"文件内容：{f.read()}")
    print(f"文件大小：{os.path.getsize(filename)} bytes")

5. 权限或表名错误 #

• 检查test_db连接使用的数据库用户是否有testdb表的插入权限；
• 确认testdb是表名而非数据库名，如果是数据库名，需要指定完整表路径（比如testdb.public.your_table）。

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