高效从DataFrame批量数据导入Redshift：优化策略与实践指南

import psycopg2
import psycopg2.extras
import pandas as pd
from io import StringIO

# 模拟一个大型DataFrame
data = {
    'case_id': range(1, 600001),
    'column_name': ['subject'] * 600000,
    'split_text': [f'text_{i}' for i in range(600000)],
    'split_text_cnt': [1] * 600000,
    'load_ts': ['2023-12-15'] * 600000
}
df = pd.DataFrame(data)

# Redshift连接信息
conn_params = {
    'host': 'redshift-####-dev.00000.us-east-1.redshift.amazonaws.com',
    'database': '*****',
    'user': '****',
    'password': '*****',
    'port': '5439'
}

try:
    conn = psycopg2.connect(**conn_params)
    cur = conn.cursor()
    print("成功连接到 Redshift Dev")

    table_name = "odey.sfc_ca_sit_di" # 替换为你的目标表名
    columns = ['case_id', 'column_name', 'split_text', 'split_text_cnt', 'load_ts']

    # 将DataFrame转换为元组列表
    data_tuples = [tuple(x) for x in df[columns].values]

    # 定义批次大小
    batch_size = 10000 # 根据实际情况调整，Redshift SQL命令最大16MB

    for i in range(0, len(data_tuples), batch_size):
        batch = data_tuples[i:i + batch_size]

        # 使用 psycopg2.extras.execute_values 进行多行插入
        # 这种方式会自动构建 VALUES (...) , (...) 的SQL语句
        psycopg2.extras.execute_values(
            cur,
            f"INSERT INTO {table_name} ({','.join(columns)}) VALUES %s",
            batch
        )
        print(f"已插入 {min(i + batch_size, len(data_tuples))} 条记录...")

    conn.commit()
    print("所有批次数据插入完成并已提交。")

except Exception as e:
    print(f"数据插入失败: {e}")
    if conn:
        conn.rollback() # 发生错误时回滚
finally:
    if cur:
        cur.close()
    if conn:
        conn.close()
    print("数据库连接已关闭。")

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

# 模拟一个大型DataFrame
data = {
    'case_id': range(1, 600001),
    'column_name': ['subject'] * 600000,
    'split_text': [f'text_{i}' for i in range(600000)],
    'split_text_cnt': [1] * 600000,
    'load_ts': ['2023-12-15'] * 600000
}
df = pd.DataFrame(data)

# Redshift连接信息
conn_params = {
    'host': 'redshift-####-dev.00000.us-east-1.redshift.amazonaws.com',
    'database': '*****',
    'user': '****',
    'password': '*****',
    'port': '5439'
}

# S3配置
s3_bucket_name = 'your-redshift-data-load-bucket' # 替换为你的S3桶名
s3_key_prefix = 'data_loads/'
s3_file_name = 'df_data_to_redshift.csv'
full_s3_path = f's3://{s3_bucket_name}/{s3_key_prefix}{s3_file_name}'

# Redshift IAM Role ARN (推荐使用IAM Role)
# 确保此IAM Role有权限访问上述S3桶
redshift_iam_role_arn = 'arn:aws:iam::YOUR_AWS_ACCOUNT_ID:role/YourRedshiftCopyRole' 

try:
    # 1. 将DataFrame保存到CSV（使用StringIO避免创建临时文件）
    csv_buffer = StringIO()
    df.to_csv(csv_buffer, index=False, header=False) # Redshift COPY通常不需要header和index

    # 2. 上传CSV数据到S3
    s3 = boto3.client('s3')
    s3.put_object(Bucket=s3_bucket_name, Key=f'{s3_key_prefix}{s3_file_name}', Body=csv_buffer.getvalue())
    print(f"数据已成功上传到 S3: {full_s3_path}")

    # 3. 连接Redshift并执行COPY命令
    conn = psycopg2.connect(**conn_params)
    cur = conn.cursor()
    print("成功连接到 Redshift Dev")

    table_name = "odey.sfc_ca_sit_di" # 替换为你的目标表名
    columns = ['case_id', 'column_name', 'split_text', 'split_text_cnt', 'load_ts']

    # 构建COPY命令
    # CSV DELIMITER ','
    # IGNOREHEADER 1 (如果你的CSV包含标题行，这里我们设置为False)
    # IAM_ROLE '...' (推荐使用IAM Role)
    # DATEFORMAT 'YYYY-MM-DD' (如果日期格式不标准)
    # TIMEFORMAT 'YYYY-MM-DD HH:MI:SS'
    # ESCAPE (处理特殊字符，如逗号在字段内)
    # REMOVEQUOTES (如果字段被双引号包围)
    # MAXERROR 允许的最大错误行数
    copy_sql = f"""
    COPY {table_name} ({','.join(columns)})
    FROM '{full_s3_path}'
    IAM_ROLE '{redshift_iam_role_arn}'
    CSV
    DELIMITER ','
    IGNOREHEADER 0  -- 因为df.to_csv(header=False)
    DATEFORMAT 'YYYY-MM-DD'
    TRUNCATECOLUMNS -- 截断超过目标列长度的字符串
    REMOVEQUOTES; -- 如果CSV字段有双引号包围
    """

    print("正在执行 Redshift COPY 命令...")
    cur.execute(copy_sql)
    conn.commit()
    print("Redshift COPY 命令执行成功，数据已加载。")

except Exception as e:
    print(f"数据加载失败: {e}")
    if conn:
        conn.rollback() # 发生错误时回滚
finally:
    if cur:
        cur.close()
    if conn:
        conn.close()
    print("数据库连接已关闭。")