如何使用Python将CSV数据导入QuestDB？psycopg2导入方法咨询

基于psycopg2模块连接QuestDB执行批量写入时，参考PostgreSQL原生copy_from用法导入CSV频繁遇到报错，需要适配QuestDB的协议兼容层规则，实现稳定的高性能CSV导入。

• 分隔符不匹配：copy_from默认采用制表符（\t）作为列分隔符，但pandas导出的CSV默认使用逗号分隔，直接读取会出现列值拼接、类型解析失败的问题
• 列不匹配：导出CSV时添加了index_label='id'的索引列，若QuestDB目标表没有对应列，会直接触发列数不匹配的错误
• 空值映射不一致：未显式声明空值的标识规则，pandas导出的空值默认是空字符串，QuestDB兼容层无法正确识别空值类型
• 资源释放逻辑不完善：临时文件、游标、连接没有在所有异常路径下正常释放，容易出现句柄占用、数据残留的问题

该方案完全复用现有psycopg2技术栈，不需要额外引入第三方SDK，导入性能可达每秒10万-20万行，相比逐行INSERT性能提升两个数量级。

修正后的可运行代码 #

import pandas as pd
import psycopg2
import os

# 初始化QuestDB连接
conn = psycopg2.connect(
    user="admin",
    password="quest",
    host="127.0.0.1",
    port="8812",
    database="qdb"
)
# 配置导入参数
dest_table = "eur_fr_bulk"
temp_csv_path = "./temp_dataframe.csv"
# 显式声明与目标表完全对齐的列名，避免顺序错位
target_columns = ["ts", "currency_pair", "bid_price", "ask_price"]

try:
    # 读取源CSV文件
    df = pd.read_csv("./data/eur_fr.csv")
    # 导出临时CSV：关闭索引导出，空值用\N对齐copy_from规则
    df.to_csv(
        temp_csv_path,
        index=False,
        na_rep=r"\N",
        header=False
    )
    # 以只读模式打开临时文件
    with open(temp_csv_path, 'r') as f:
        cursor = conn.cursor()
        # 显式传入与QuestDB兼容的copy_from参数
        cursor.copy_from(
            file=f,
            table=dest_table,
            sep=',',
            null=r'\N',
            columns=target_columns
        )
        conn.commit()
        cursor.close()
except Exception as e:
    # 异常场景下回滚事务
    conn.rollback()
    print(f"导入执行出错：{str(e)}")
finally:
    # 全场景下清理资源
    if os.path.exists(temp_csv_path):
        os.remove(temp_csv_path)
    conn.close()

落地注意事项 #

• 导入前确认QuestDB目标表的列顺序、数据类型与target_columns完全对齐，时序表必须包含TIMESTAMP类型的主键时间列
• 若CSV包含时间字段，建议提前用pandas将时间列转换为UTC时区的datetime64类型再导出，避免QuestDB对字符串时间的隐式解析错误
• 当单次导入数据量超过100万行时，建议拆分CSV为10万行粒度的块分批导入，避免长事务导致的连接超时

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