如何在Redshift的COPY命令中添加versionid与load_timestamp额外列？

方法一：临时表中转（兼容性强，适配所有Redshift版本） #

1. 目标表结构准备 #

先确保Redshift目标表包含CSV原有列，同时新增versionid和load_timestamp字段：

CREATE TABLE target_table (
  -- 以下为CSV原有列，请根据实际结构调整
  csv_col1 VARCHAR(255),
  csv_col2 INT,
  csv_col3 DATE,
  -- 额外添加的字段
  versionid VARCHAR(100),
  load_timestamp TIMESTAMP
);

2. Lambda执行的SQL逻辑 #

Lambda触发时，从S3事件中提取对象的versionId（路径：event.Records[0].s3.object.versionId），然后执行以下SQL序列：

-- 创建临时 staging 表，结构与CSV完全匹配（不含额外字段）
CREATE TEMP TABLE temp_staging (
  csv_col1 VARCHAR(255),
  csv_col2 INT,
  csv_col3 DATE
);

-- 将S3中的CSV数据导入临时表
COPY temp_staging
FROM 's3://your-bucket-name/path/to/your/file.csv'
IAM_ROLE 'arn:aws:iam::你的账号ID:role/Redshift访问S3的角色ARN'
FORMAT AS CSV
DELIMITER ','
IGNOREHEADER 1; -- 如果CSV包含表头，忽略第一行

-- 将临时表数据插入目标表，同时填充额外字段
INSERT INTO target_table (csv_col1, csv_col2, csv_col3, versionid, load_timestamp)
SELECT 
  csv_col1, 
  csv_col2, 
  csv_col3, 
  '从Lambda获取的versionId值', -- 替换为实际拿到的versionId
  CURRENT_TIMESTAMP -- 自动生成加载时间
FROM temp_staging;

-- 清理临时表
DROP TABLE temp_staging;

方法二：直接使用COPY元数据函数（更简洁，需Redshift 1.0.2367及以上版本） #

如果你的Redshift版本支持GET_METADATA函数，可跳过临时表，直接在COPY语句中填充额外字段：

1. 目标表结构同方法一 #

2. COPY语句示例 #

COPY target_table (csv_col1, csv_col2, csv_col3, versionid, load_timestamp)
FROM 's3://your-bucket-name/path/to/your/file.csv'
IAM_ROLE 'arn:aws:iam::你的账号ID:role/Redshift访问S3的角色ARN'
FORMAT AS CSV
DELIMITER ','
IGNOREHEADER 1
-- 映射额外字段：从S3对象元数据取versionid，用当前时间作为加载时间
COLUMNS (
  csv_col1, 
  csv_col2, 
  csv_col3, 
  versionid GET_METADATA('$PATH', 'version_id'),
  load_timestamp AS CURRENT_TIMESTAMP
);

关键注意事项 #

• Lambda权限：需配置IAM权限，允许读取S3对象版本信息（s3:GetObjectVersion），以及访问Redshift的权限（如Redshift Data API权限或JDBC连接权限）。
• 批量处理：若Lambda一次收到多个S3事件，需循环遍历每个事件，分别提取对应versionId并执行导入逻辑。
• 错误处理：在Lambda中添加异常捕获，记录失败日志，必要时触发告警或重试机制。

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