导入S3文件至Redshift时如何保留原文件行号？

方法1：使用COPY命令结合ROW_NUMBER()与文件排序 #

Redshift的COPY命令默认是并行加载的，但如果你的S3文件有可识别的序列规则（比如文件名按data_001.csv、data_002.csv这类顺序命名），可以通过以下步骤生成对应原文件顺序的行号：

1. 用COPY命令将数据加载到临时表，同时通过$path伪列保留文件路径信息：

COPY staging_table
FROM 's3://your-bucket/data-prefix/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftLoadRole'
FORMAT AS CSV;

2. 利用窗口函数按文件顺序+文件内行顺序生成全局行号，插入目标表：

INSERT INTO target_table (row_num, col1, col2, ...)
SELECT 
  ROW_NUMBER() OVER (ORDER BY s3_path, internal_row_id) AS row_num,
  col1, col2, ...
FROM (
  SELECT 
    $path AS s3_path,
    -- 单文件内按物理行顺序生成子行号（未压缩CSV通常能保证顺序）
    ROW_NUMBER() OVER (PARTITION BY $path ORDER BY col1) AS internal_row_id,
    col1, col2, ...
  FROM staging_table
) t;

注：如果文件内没有天然的顺序标识列，单文件的行顺序依赖Redshift加载时的稳定性，未压缩、格式规整的文件通常能保持原物理顺序。

方法2：用Manifest文件强制加载顺序 #

如果S3文件的命名序列明确，可以创建JSON格式的Manifest文件，按原数据的行顺序列出所有文件路径，以此控制COPY的加载顺序：

1. 生成Manifest文件示例：

{
  "entries": [
    {"url": "s3://your-bucket/data_001.csv"},
    {"url": "s3://your-bucket/data_002.csv"},
    {"url": "s3://your-bucket/data_003.csv"}
  ]
}

2. 将Manifest上传到S3，用COPY加载临时表：

COPY staging_table
FROM 's3://your-bucket/manifest.json'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftLoadRole'
FORMAT AS CSV
MANIFEST;

3. 基于Manifest的文件顺序生成行号：

INSERT INTO target_table (row_num, col1, col2, ...)
SELECT
  ROW_NUMBER() OVER (ORDER BY manifest_order, internal_row_id) AS row_num,
  col1, col2, ...
FROM (
  SELECT
    -- 按Manifest中的文件顺序排序
    ROW_NUMBER() OVER (ORDER BY $path) AS manifest_order,
    ROW_NUMBER() OVER (PARTITION BY $path ORDER BY col1) AS internal_row_id,
    col1, col2, ...
  FROM staging_table
) t;

方法3：用AWS Glue ETL动态添加行号（不修改原文件） #

通过Glue Job读取S3文件时，直接为每行生成对应原顺序的行号，再写入Redshift：

• 示例Python脚本片段：

from pyspark.sql.functions import row_number, monotonically_increasing_id
from pyspark.sql.window import Window

# 读取S3文件
df = spark.read.csv("s3://your-bucket/data-prefix/", header=True)

# 按文件路径+文件内行顺序生成连续行号
window_spec = Window.orderBy("input_file_name", monotonically_increasing_id())
df_with_row_num = df.withColumn("row_num", row_number().over(window_spec))

# 写入Redshift
df_with_row_num.write \
  .format("jdbc") \
  .option("url", "jdbc:redshift://your-cluster-url:5439/your-db") \
  .option("dbtable", "target_table") \
  .option("user", "your-user") \
  .option("password", "your-password") \
  .option("tempdir", "s3://your-bucket/temp/") \
  .mode("append") \
  .save()

这个方案不需要修改原S3文件，完全通过ETL过程在加载时生成行号。

补充：为什么Spectrum+Identity列会顺序混乱？ #

Redshift Spectrum是并行扫描S3数据的，查询时的并行处理会打乱原文件的行顺序；而IDENTITY列是按Redshift插入数据的顺序生成的，和原文件的行顺序没有对应关系，因此会出现顺序错乱的情况。

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