如何将S3存储桶20230405目录下的100+CSV文件合并为result.csv

以下是几种实用的解决方案，按需选择：

方法一：使用AWS CLI（适合快速操作） #

如果文件总大小不大，可通过CLI快速处理：

• 方案1：本地合并后上传

  1. 将S3目录下的CSV同步到本地临时目录：

     aws s3 cp s3://your-bucket/20230405/ ./temp-csv/ --recursive --exclude "*" --include "*.csv"
     

  2. 合并文件，仅保留首个文件的表头：

     head -n 1 ./temp-csv/$(ls ./temp-csv/ | head -n 1) > result.csv && tail -q -n +2 ./temp-csv/*.csv >> result.csv
     

  3. 将合并后的文件上传回S3：

     aws s3 cp result.csv s3://your-bucket/20230405/result.csv
     

  4. 清理本地临时文件：

     rm -rf ./temp-csv/ result.csv
     

• 方案2：直接云端读取合并（无需本地存储）
  适合不想下载文件的场景，通过管道直接处理内容：

  aws s3 cp s3://your-bucket/20230405/ - --recursive --exclude "*" --include "*.csv" | awk 'NR==1 || FNR>1' > result.csv
  aws s3 cp result.csv s3://your-bucket/20230405/result.csv
  

方法二：使用Python（boto3）（适合自定义处理逻辑） #

如果需要对CSV内容做清洗、转换等操作，用Python脚本更灵活：

import boto3
import csv
from io import StringIO

# 配置参数
bucket_name = "your-bucket"
prefix = "20230405/"

s3 = boto3.client("s3")

# 获取目标目录下所有CSV文件（排除已生成的result.csv）
response = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
csv_files = [
    obj["Key"] for obj in response.get("Contents", [])
    if obj["Key"].endswith(".csv") and obj["Key"] != f"{prefix}result.csv"
]

output_buffer = StringIO()
csv_writer = csv.writer(output_buffer)
header_written = False

for file_key in csv_files:
    # 读取S3文件内容
    obj = s3.get_object(Bucket=bucket_name, Key=file_key)
    content = obj["Body"].read().decode("utf-8")
    csv_reader = csv.reader(StringIO(content))

    if not header_written:
        # 写入第一个文件的表头
        csv_writer.writerows(csv_reader)
        header_written = True
    else:
        # 跳过表头，写入数据行
        next(csv_reader)
        csv_writer.writerows(csv_reader)

# 将合并结果上传回S3
s3.put_object(
    Bucket=bucket_name,
    Key=f"{prefix}result.csv",
    Body=output_buffer.getvalue().encode("utf-8")
)

print("合并完成，result.csv已上传至指定S3路径")

注意：需先安装boto3（pip install boto3），并配置好AWS访问凭证。

方法三：使用Amazon Athena（适合超大规模文件） #

无需本地资源，直接在云端处理超大文件集合：

1. 创建Athena外部表（根据你的CSV字段结构调整定义）：

   CREATE EXTERNAL TABLE IF NOT EXISTS csv_data (
       col1 string,
       col2 int,
       col3 timestamp
       -- 按需添加更多字段
   )
   ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
   WITH SERDEPROPERTIES (
       'serialization.format' = ',',
       'field.delim' = ','
   )
   LOCATION 's3://your-bucket/20230405/'
   TBLPROPERTIES ('has_encrypted_data'='false', 'skip.header.line.count'='1');
   

2. 导出数据到S3：

   INSERT OVERWRITE DIRECTORY 's3://your-bucket/20230405/result_temp/'
   ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
   WITH SERDEPROPERTIES (
       'serialization.format' = ',',
       'field.delim' = ','
   )
   SELECT * FROM csv_data;
   

3. 合并Athena生成的分片文件：

   aws s3 cp s3://your-bucket/20230405/result_temp/ - --recursive | awk 'NR==1 || FNR>1' > result.csv
   aws s3 cp result.csv s3://your-bucket/20230405/result.csv
   

注意：Athena输出目录需与源目录分离，避免冲突。

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