Java实现基于类别拆分S3存储的每日大型CSV文件

嘿，这个问题我刚好处理过类似的场景，咱们既要保持流式读取不炸内存，又要按类别精准拆分CSV，核心思路就是维护一个「类别→写入流」的映射，同时做好资源管理避免踩坑。我给你拆解下具体步骤和示例代码：

因为是流式逐行处理，我们不需要把整个大文件加载到内存里，而是每读取一行就判断它的类别，然后把它写入对应类别的文件中。同时要注意：每个类别文件需要先写入CSV表头，还要避免打开过多文件句柄导致系统报错，以及处理完后要正确关闭所有资源。

1. 先提取CSV表头：
   流式读取文件的第一行（表头），之后每个新类别的文件创建时，先写入这行表头，确保每个拆分后的小文件结构完整。

2. 建立类别到写入流的映射：

   • 用S3的流式API读取大文件（比如Boto3的get_object返回的Body对象支持逐行迭代）
   • 逐行解析CSV，提取每行的类别字段值
   • 检查该类别是否已有对应的写入流：
     • 如果没有，创建临时文件（或直接准备S3的写入流），写入表头，然后把当前行写入
     • 如果已有，直接把当前行写入对应的流

3. 资源收尾工作：

   • 所有行处理完成后，关闭所有打开的文件句柄
   • 把临时文件上传到S3的指定路径（比如按类别命名的对象路径）
   • 清理本地临时文件（避免磁盘占用）

下面是一个可直接参考的实现，用临时目录存放中间文件，避免内存溢出：

import csv
import boto3
from tempfile import TemporaryDirectory
from collections import defaultdict

def split_large_s3_csv_by_category(s3_bucket, s3_input_key, category_field, s3_output_prefix):
    # 初始化S3客户端
    s3_client = boto3.client('s3')
    
    # 获取S3对象的流式迭代器，按块读取避免内存压力
    s3_response = s3_client.get_object(Bucket=s3_bucket, Key=s3_input_key)
    csv_line_iterator = s3_response['Body'].iter_lines(chunk_size=1024*1024, decode_unicode=True)
    
    # 初始化CSV字典读取器，自动解析表头
    csv_reader = csv.DictReader(csv_line_iterator)
    csv_header = csv_reader.fieldnames
    
    # 维护类别到(文件句柄, CSV写入器)的映射
    category_handlers = defaultdict(lambda: None)
    
    # 使用临时目录存放中间文件，自动清理
    with TemporaryDirectory() as temp_work_dir:
        try:
            for row_num, row in enumerate(csv_reader, start=2):  # 从第2行开始计数（表头是第1行）
                current_category = row.get(category_field)
                if not current_category:
                    # 处理无类别字段的行，可以跳过或写入"unknown"类别
                    current_category = "unknown"
                
                if category_handlers[current_category] is None:
                    # 为新类别创建临时文件和写入器
                    temp_file_path = f"{temp_work_dir}/{current_category}.csv"
                    file_handle = open(temp_file_path, 'w', newline='', encoding='utf-8')
                    csv_writer = csv.writer(file_handle)
                    # 写入表头
                    csv_writer.writerow(csv_header)
                    category_handlers[current_category] = (file_handle, csv_writer)
                
                # 写入当前行到对应类别的文件
                file_handle, csv_writer = category_handlers[current_category]
                csv_writer.writerow(row.values())
                
                # 可选：每处理10000行打印进度
                if row_num % 10000 == 0:
                    print(f"Processed {row_num} rows...")
            
            # 所有行处理完成，上传所有类别文件到S3
            for category, (file_handle, _) in category_handlers.items():
                # 先关闭文件句柄，确保内容写入磁盘
                file_handle.close()
                # 构造S3输出路径
                s3_output_key = f"{s3_output_prefix}/{category}.csv"
                # 上传文件到S3
                s3_client.upload_file(f"{temp_work_dir}/{category}.csv", s3_bucket, s3_output_key)
                print(f"Successfully uploaded {category} data to s3://{s3_bucket}/{s3_output_key}")
        
        finally:
            # 确保所有文件句柄都关闭，即使中途出错
            for category, (file_handle, _) in category_handlers.items():
                if not file_handle.closed:
                    file_handle.close()
            print("All resources cleaned up.")

# 调用示例，替换成你的实际参数
split_large_s3_csv_by_category(
    s3_bucket="your-target-bucket",
    s3_input_key="path/to/large-input.csv",
    category_field="your-category-column-name",
    s3_output_prefix="split-by-category"
)

• 文件句柄限制：如果你的CSV里有上万种不同类别，可能会触发系统的文件句柄上限。这时候可以考虑设置一个阈值，定期关闭不活跃的文件句柄，后续再遇到该类别时重新打开（记得不要重复写入表头）。
• 编码一致性：确保CSV文件的编码和代码中指定的编码一致（比如utf-8），避免出现乱码。
• 错误重试：S3上传可能会遇到网络问题，建议给上传步骤加上重试机制（比如用tenacity库实现）。
• 临时文件清理：示例中用了TemporaryDirectory，会自动清理临时文件，避免磁盘被占满。
• 表头唯一性：确保每个类别文件只写入一次表头，示例中在创建新类别文件时就写入表头，避免重复。

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