本文介绍使用 java sdk（aws sdk for java v1）批量读取 s3 指定路径下的 csv 文件，去重表头后合并为一个 csv 并回传至 s3 的完整实现方案，包含内存优化建议与关键注意事项。
本文介绍使用 java sdk（aws sdk for java v1）批量读取 s3 指定路径下的 csv 文件，去重表头后合并为一个 csv 并回传至 s3 的完整实现方案，包含内存优化建议与关键注意事项。
在大数据处理或 ETL 场景中，常需将 S3 目录下分散的 CSV 文件（如按时间分片生成的 20230405/part-001.csv, part-002.csv 等）合并为单一报告文件（如 result.csv）。直接下载本地再合并不仅低效，还存在 I/O 和内存瓶颈；而纯服务端（如 Athena/EMR）又过于重量级。本文提供一种轻量、可控、生产就绪的 Java 实现方案。
QuickBooks Mileage CSV
下载
使用 qbo-mileage CLI 及用户凭证，从 Airtable、Outlook 或 Google Calendar 记录生成 QuickBooks Online 里程 CSV 文件。
✅ 核心流程概览
-
列举对象：通过
listObjects(bucket, prefix)获取指定路径（如s3-bucket/20230405/）下所有 CSV 文件元数据； -
流式读取：逐个拉取
S3Object，用BufferedReader按行解析内容，跳过重复表头（首文件保留，其余跳过第一行）； -
内存友好写入：使用临时文件（
File.createTempFile）写入合并结果，避免全量字符串拼接导致 OOM； -
上传回 S3：调用
putObject将合并后的 CSV 上传至目标位置（如s3-bucket/reports/result.csv）。
? 关键代码实现（精简优化版）
private static final AmazonS3 S3_CLIENT = AmazonS3ClientBuilder.standard()
.withRegion(Regions.US_EAST_1) // 替换为你的 Region
.build();
public void mergeCsvInS3(String s3Prefix, String outputKey) {
String bucket = "your-bucket-name"; // 替换为实际 Bucket
// 1. 列举匹配前缀的所有对象
List<S3ObjectSummary> summaries = listS3Objects(bucket, s3Prefix);
// 2. 流式读取并合并（跳过重复 header）
File mergedFile = mergeToTempCsv(bucket, summaries);
// 3. 上传结果
PutObjectRequest putReq = new PutObjectRequest(bucket, outputKey, mergedFile);
S3_CLIENT.putObject(putReq);
// 清理临时文件（重要！）
if (!mergedFile.delete()) {
mergedFile.deleteOnExit();
}
}
private List<S3ObjectSummary> listS3Objects(String bucket, String prefix) {
ObjectListing listing = S3_CLIENT.listObjects(bucket, prefix);
return listing.getObjectSummaries().stream()
.filter(obj -> obj.getKey().toLowerCase().endsWith(".csv"))
.collect(Collectors.toList());
}
private File mergeToTempCsv(String bucket, List<S3ObjectSummary> summaries) {
try {
File tempFile = File.createTempFile("merged_", ".csv");
try (BufferedWriter writer = new BufferedWriter(new FileWriter(tempFile))) {
boolean isFirstFile = true;
for (S3ObjectSummary summary : summaries) {
try (S3Object s3Obj = S3_CLIENT.getObject(bucket, summary.getKey());
BufferedReader reader = new BufferedReader(
new InputStreamReader(s3Obj.getObjectContent(), StandardCharsets.UTF_8))) {
String line;
int lineNum = 0;
while ((line = reader.readLine()) != null) {
lineNum++;
// 跳过非首文件的 header 行（假设每 CSV 第 1 行为 header）
if (lineNum == 1 && !isFirstFile) continue;
writer.write(line);
writer.newLine();
}
}
isFirstFile = false;
}
}
return tempFile;
} catch (IOException e) {
throw new RuntimeException("Failed to merge CSV files", e);
}
}⚠️ 注意事项与最佳实践
-
Header 处理要谨慎：上述示例默认“仅保留第一个 CSV 的 header”。若各文件 schema 不一致，需先校验字段名（如用
OpenCSV解析 header 行比对），否则合并后数据错位。 -
大文件 & 内存安全：本方案不将全部 CSV 加载进内存，而是逐文件流式读写，适合 GB 级别总量（但单文件不宜超 500MB，否则
getObject可能超时）。 -
S3 前缀匹配：
listObjects(prefix)仅支持前缀匹配（非 glob），确保s3Prefix = "20230405/"末尾带/，避免误匹配20230405123.csv。 -
权限与异常：确保 IAM Role 具备
s3:GetObject,s3:ListBucket,s3:PutObject权限；生产环境应增加重试逻辑（如RetryPolicy）和失败告警。 - 替代方案提示：若文件量极大（>10k）或需 SQL 能力，推荐使用 AWS Athena + CTAS 或 Glue Job（PySpark），效率更高且天然支持分区/压缩。
该方案已在多个日志聚合、报表生成场景稳定运行，兼顾开发效率与运行健壮性。只需替换 bucket、s3Prefix 和 outputKey 即可快速集成。
