如何通过psycopg2 copy_expert高效导入CSV至PostgreSQL并处理列需求

需求说明 #

• 原CSV含6列（示例为timer;A1;A2;A3;A4;UnitMeasureent），仅需导入其中4列（跳过timer和A3）
• 导入列需重命名：CSV的A1→表的a1，A2→a2，A4→a4，UnitMeasureent→unit_measurement
• 需添加两个静态值列：serial_number固定为S01，global_id固定为GER
• 替换逐行INSERT的低效方式，用copy_expert实现高速导入

解决方案（PostgreSQL 12+） #

利用PostgreSQL的COPY FROM STDIN配合AS子句，直接在导入时处理列筛选、重命名和静态值添加，无需临时表。

Python代码示例

import psycopg2

# 替换为你的数据库连接参数
conn_params = {
    "dbname": "your_database",
    "user": "your_username",
    "password": "your_password",
    "host": "your_host",
    "port": "5432"
}

# 定义COPY命令，处理列映射和静态值
copy_sql = """
COPY public.data (serial_number, global_id, a1, a2, a4, unit_measurement)
FROM STDIN
WITH (FORMAT csv, DELIMITER ';', HEADER true)
AS (
    SELECT 
        'S01'::varchar,
        'GER'::varchar,
        $2::double precision,  -- CSV第2列A1映射到表a1
        $3::double precision,  -- CSV第3列A2映射到表a2
        $5::double precision,  -- CSV第5列A4映射到表a4
        $6::varchar(50)        -- CSV第6列UnitMeasureent映射到表unit_measurement
);
"""

# 执行导入
with psycopg2.connect(**conn_params) as conn:
    with conn.cursor() as cur:
        # 二进制模式读取CSV，避免换行符兼容问题
        with open("your_csv_file.csv", "rb") as csv_file:
            cur.copy_expert(copy_sql, csv_file)
    conn.commit()

兼容低版本PostgreSQL（<12） #

如果你的PostgreSQL版本低于12，不支持COPY的AS子句，可以先导入到临时表，再通过INSERT语句处理列映射和静态值：

Python代码示例

import psycopg2

conn_params = {
    "dbname": "your_database",
    "user": "your_username",
    "password": "your_password",
    "host": "your_host",
    "port": "5432"
}

# 创建临时表→导入CSV→插入目标表→清理临时表的组合SQL
copy_sql = """
CREATE TEMP TABLE temp_csv (
    timestamp timestamp,
    A1 double precision,
    A2 double precision,
    A3 double precision,
    A4 double precision,
    UnitMeasureent varchar(50)
);

COPY temp_csv FROM STDIN WITH (FORMAT csv, DELIMITER ';', HEADER true);

INSERT INTO public.data (serial_number, global_id, a1, a2, a4, unit_measurement)
SELECT 'S01', 'GER', A1, A2, A4, UnitMeasureent FROM temp_csv;

DROP TABLE temp_csv;
"""

with psycopg2.connect(**conn_params) as conn:
    with conn.cursor() as cur:
        with open("your_csv_file.csv", "rb") as csv_file:
            cur.copy_expert(copy_sql, csv_file)
    conn.commit()

关键说明 #

• copy_expert直接调用PostgreSQL原生COPY命令，导入速度远高于逐行INSERT
• HEADER true自动跳过CSV的第一行表头
• 二进制模式读取CSV避免不同操作系统换行符的解析问题
• 类型转换（如::double precision）确保数据类型与目标表匹配

内容的提问来源于stack exchange，提问作者Pierre-Alexandre