如何提升Python Pandas向PieCloudDB导入数据的效率？

针对25万条数据的导入场景，以下是几个实用的优化方法：

• 用COPY协议替代INSERT批量写入
  PieCloudDB兼容PostgreSQL生态，COPY是原生的高速导入方式，比to_sql的multi方法效率高一个量级。可以通过psycopg2的copy_from实现：

import psycopg2
from io import StringIO
import pandas as pd
import datetime

def copy_to_pieclouddb(df: pd.DataFrame, table="t_weather_history"):
    start_time = datetime.datetime.now()
    # 替换为你的数据库连接参数
    conn = psycopg2.connect(
        dbname="your_db",
        user="your_user",
        password="your_pwd",
        host="your_host",
        port="your_port"
    )
    cur = conn.cursor()
    # 用内存缓冲区模拟CSV文件
    output = StringIO()
    df.to_csv(output, sep='\t', header=False, index=False, na_rep='')
    output.seek(0)
    # 执行COPY命令
    cur.copy_from(output, table, sep='\t', null='')
    conn.commit()
    
    duration = (datetime.datetime.now() - start_time).total_seconds()
    print(f"COPY导入{len(df)}条数据耗时{duration:.2f}秒")
    
    cur.close()
    conn.close()

• 优化to_sql的批量配置
  如果坚持用to_sql，调整chunksize参数并通过事务包裹操作，减少提交开销：

import pandas as pd
import datetime
from sqlalchemy import create_engine

def insert_rows(rows: pd.DataFrame, table="t_weather_history", conn=None, commit_every=50000):
    if len(rows) == 0:
        return
    start_time = datetime.datetime.now()
    
    if conn is None:
        engine = create_engine("postgresql+psycopg2://user:pwd@host:port/dbname")
        # 用事务包裹，避免频繁提交
        with engine.begin() as conn:
            rows.to_sql(
                table,
                con=conn,
                if_exists='append',
                index=False,
                method='multi',
                chunksize=commit_every
            )
    else:
        with conn.begin():
            rows.to_sql(
                table,
                con=conn,
                if_exists='append',
                index=False,
                method='multi',
                chunksize=commit_every
            )
    
    duration = (datetime.datetime.now() - start_time).total_seconds()
    print(f"插入{len(rows)}条数据到表{table}耗时{duration:.2f}秒")

建议把chunksize设为5万左右，平衡单次提交的数据量和网络交互次数。

• 临时禁用索引与约束
  导入前先关闭目标表的索引、外键约束和触发器，导入完成后再恢复——这些约束会在每条数据插入时触发校验/更新，严重拖慢大批次导入速度：

# 导入前执行，关闭约束
def disable_constraints(engine, table="t_weather_history"):
    with engine.connect() as conn:
        conn.execute(f"ALTER TABLE {table} DISABLE TRIGGER ALL;")
        # 替换为你的索引名，提前删除导入后重建
        conn.execute(f"DROP INDEX IF EXISTS idx_weather_time;")
        conn.commit()

# 导入后执行，恢复约束
def enable_constraints(engine, table="t_weather_history"):
    with engine.connect() as conn:
        conn.execute(f"ALTER TABLE {table} ENABLE TRIGGER ALL;")
        conn.execute(f"CREATE INDEX idx_weather_time ON {table}(record_time);")
        conn.commit()

• 调优数据库连接参数
  创建SQLAlchemy引擎时，增大连接池尺寸、关闭自动提交，减少连接建立的开销：

def get_sqlalchemy_engine():
    conn_str = "postgresql+psycopg2://user:pwd@host:port/dbname"
    return create_engine(
        conn_str,
        pool_size=20,  # 基础连接数
        max_overflow=30,  # 最大额外连接数
        autocommit=False,
        client_encoding='utf8'
    )

• 对齐数据类型
  提前把DataFrame中的数据类型和数据库表字段类型对齐，比如把Python的datetime64转换成数据库的timestamp，字符串长度匹配字段定义，避免Pandas在写入时自动转换带来的额外开销。

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