如何高效将多份Feather文件批量导入PostgreSQL并自动建表？

问题背景 #

我有多个Feather文件，每个文件包含约120列、30万条数据。已经写了Python代码自动生成PostgreSQL建表语句，但不知道怎么高效按列名导入数据。现有代码如下：

path = "E:\file_path"
list = glob.glob(os.path.join(path, '*.feather'))

for file in list:
    print(f"File: {file}")
    df = pd.read_feather(file)
    table_name = {file}

    columns = df.dtypes.to_dict()
    columns_sql = []

    def map_dtype_to_pg(dtype): # this function is for changing to the type input in sql
        if pd.api.types.is_integer_dtype(dtype):
            return "INTEGER"
        elif pd.api.types.is_float_dtype(dtype):
            return "FLOAT"
        elif pd.api.types.is_bool_dtype(dtype):
            return "BOOLEAN"
        elif pd.api.types.is_datetime64_any_dtype(dtype):
            return "TIMESTAMP"
        else:
            return "TEXT"

    for col, dtype in columns.items(): # the column's name and type will be recorded here
        pg_type = map_dtype_to_pg(dtype)
        columns_sql.append(f'"{col}" {pg_type}')

    create_table_sql = f"CREATE TABLE {table_name} (  {' '.join(columns_sql)}\n);" # this will be the query to create multiple table from the existing file's name and value

高效导入方案 #

方法1：Pandas to_sql 结合批量插入（推荐） #

默认to_sql逐行插入速度极慢，换成method='multi'批量提交能大幅提速，同时保留自动匹配列名的优势。先执行建表逻辑，再导入数据：

import pandas as pd
from sqlalchemy import create_engine
import glob
import os

# 数据库连接（替换成你的信息）
engine = create_engine('postgresql://用户名:密码@主机:端口/数据库名')

path = "E:\\file_path"
file_list = glob.glob(os.path.join(path, '*.feather'))

for file in file_list:
    print(f"处理文件: {file}")
    df = pd.read_feather(file)
    # 从文件名提取合法表名（去掉路径和后缀）
    table_name = os.path.splitext(os.path.basename(file))[0]
    
    # 执行建表逻辑（新增IF NOT EXISTS避免重复建表）
    columns = df.dtypes.to_dict()
    columns_sql = []

    def map_dtype_to_pg(dtype):
        if pd.api.types.is_integer_dtype(dtype):
            return "INTEGER"
        elif pd.api.types.is_float_dtype(dtype):
            return "FLOAT"
        elif pd.api.types.is_bool_dtype(dtype):
            return "BOOLEAN"
        elif pd.api.types.is_datetime64_any_dtype(dtype):
            return "TIMESTAMP"
        else:
            return "TEXT"

    for col, dtype in columns.items():
        pg_type = map_dtype_to_pg(dtype)
        columns_sql.append(f'"{col}" {pg_type}')

    create_table_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join(columns_sql)});"
    with engine.connect() as conn:
        conn.execute(create_table_sql)
        conn.commit()
    
    # 批量导入数据，chunksize可根据内存调整
    df.to_sql(
        name=table_name,
        con=engine,
        if_exists='append',
        index=False,
        method='multi',
        chunksize=1000
    )

方法2：Psycopg2 copy_from（更快，适合大数据） #

基于PostgreSQL原生COPY命令，速度比to_sql更快。把DataFrame转成内存CSV对象后直接导入：

import pandas as pd
import psycopg2
from io import StringIO
import glob
import os

# 数据库连接（替换成你的信息）
conn = psycopg2.connect(
    dbname='数据库名',
    user='用户名',
    password='密码',
    host='主机',
    port='端口'
)
cur = conn.cursor()

path = "E:\\file_path"
file_list = glob.glob(os.path.join(path, '*.feather'))

for file in file_list:
    print(f"处理文件: {file}")
    df = pd.read_feather(file)
    table_name = os.path.splitext(os.path.basename(file))[0]
    
    # 建表逻辑（同方法1）
    columns = df.dtypes.to_dict()
    columns_sql = []

    def map_dtype_to_pg(dtype):
        if pd.api.types.is_integer_dtype(dtype):
            return "INTEGER"
        elif pd.api.types.is_float_dtype(dtype):
            return "FLOAT"
        elif pd.api.types.is_bool_dtype(dtype):
            return "BOOLEAN"
        elif pd.api.types.is_datetime64_any_dtype(dtype):
            return "TIMESTAMP"
        else:
            return "TEXT"

    for col, dtype in columns.items():
        pg_type = map_dtype_to_pg(dtype)
        columns_sql.append(f'"{col}" {pg_type}')

    create_table_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join(columns_sql)});"
    cur.execute(create_table_sql)
    conn.commit()
    
    # 将DataFrame转为制表符分隔的内存CSV
    buffer = StringIO()
    df.to_csv(buffer, index=False, header=False, sep='\t')
    buffer.seek(0)
    
    # 按列名导入数据
    cur.copy_from(
        file=buffer,
        table=table_name,
        sep='\t',
        columns=tuple(df.columns)
    )
    conn.commit()

# 关闭连接
cur.close()
conn.close()

方法3：导出CSV后用PostgreSQL COPY命令（超大数据首选） #

如果文件体量极大，先转成CSV再用原生COPY命令，这是最快的方式：

1. Feather转CSV：

import pandas as pd
import glob
import os

path = "E:\\file_path"
file_list = glob.glob(os.path.join(path, '*.feather'))

for file in file_list:
    df = pd.read_feather(file)
    csv_file = os.path.splitext(file)[0] + '.csv'
    df.to_csv(csv_file, index=False, sep='\t')

2. 在PostgreSQL执行导入（psql或Python执行均可）：

COPY 表名 FROM 'E:\file_path\目标文件.csv' WITH (FORMAT CSV, HEADER, DELIMITER '\t');

关键注意事项 #

• 表名修正：原代码中table_name = {file}会生成集合，必须改成从文件名提取合法表名（比如去掉特殊字符）
• 数据类型适配：如果有大整数，可把INTEGER换成BIGINT避免溢出
• 批量大小：chunksize建议在1000-10000之间，根据内存调整
• 事务控制：导入时用事务包裹，避免中途出错导致数据不全

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