如何在Java中逐行将S3中的CSV文件写入DynamoDB

1. 引入依赖 #

首先添加OpenCSV的Maven依赖，用于解析带分隔符的CSV文件：

<dependency>
    <groupId>com.opencsv</groupId>
    <artifactId>opencsv</artifactId>
    <version>5.6</version> <!-- 使用最新稳定版本即可 -->
</dependency>

2. 核心转换与写入代码 #

结合你已有的S3客户端和DynamoDB Mapper，实现完整的处理逻辑：

import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBMapper;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.S3Object;
import com.opencsv.bean.CsvToBeanBuilder;
import com.opencsv.bean.HeaderColumnNameMappingStrategy;

import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.List;

public class S3CsvToDynamoDBProcessor {

    private final AmazonS3 s3Client;
    private final DynamoDBMapper ddbMapper;

    public S3CsvToDynamoDBProcessor(AmazonS3 s3Client, DynamoDBMapper ddbMapper) {
        this.s3Client = s3Client;
        this.ddbMapper = ddbMapper;
    }

    public void processCsv(String s3Bucket, String s3Key) {
        // 获取S3对象并打开输入流，用try-with-resources自动关闭资源
        try (S3Object s3Object = s3Client.getObject(new GetObjectRequest(s3Bucket, s3Key));
             InputStreamReader reader = new InputStreamReader(s3Object.getObjectContent(), StandardCharsets.UTF_8)) {

            // 配置CSV映射策略：只匹配StatusObject的字段，自动忽略多余列
            HeaderColumnNameMappingStrategy<StatusObject> mappingStrategy = new HeaderColumnNameMappingStrategy<>();
            mappingStrategy.setType(StatusObject.class);
            mappingStrategy.setColumnSeparator('|'); // 指定CSV分隔符为|
            mappingStrategy.setTrim(true); // 自动去除表头和字段值前后的空格

            // 解析CSV为StatusObject列表
            List<StatusObject> statusObjects = new CsvToBeanBuilder<StatusObject>(reader)
                    .withMappingStrategy(mappingStrategy)
                    .build()
                    .parse();

            // 逐行写入DynamoDB（也可根据需求改为批量写入）
            for (StatusObject obj : statusObjects) {
                ddbMapper.save(obj); // 替换你之前的load方法，save用于写入/更新数据
            }

        } catch (Exception e) {
            throw new RuntimeException("处理CSV文件失败", e);
        }
    }
}

3. StatusObject类的注解配置 #

确保你的StatusObject类同时适配DynamoDB和OpenCSV的映射规则，若表头与字段名一致，@CsvBindByName可省略：

import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBAttribute;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBHashKey;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBTable;
import com.opencsv.bean.CsvBindByName;

@DynamoDBTable(tableName = "YourStatusTableName") // 替换为你的DynamoDB表名
public class StatusObject {

    @DynamoDBHashKey
    @CsvBindByName(column = "id")
    private String id;

    @DynamoDBAttribute
    @CsvBindByName(column = "name")
    private String name;

    @DynamoDBAttribute
    @CsvBindByName(column = "age")
    private Integer age;

    @DynamoDBAttribute
    @CsvBindByName(column = "height")
    private Double height;

    // 必须保留无参构造函数，DynamoDBMapper和OpenCSV都依赖它
    public StatusObject() {}

    // Getter和Setter方法
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }

    public String getName() { return name; }
    public void setName(String name) { this.name = name; }

    public Integer getAge() { return age; }
    public void setAge(Integer age) { this.age = age; }

    public Double getHeight() { return height; }
    public void setHeight(Double height) { this.height = height; }
}

关键细节说明 #

• 自动忽略多余列：HeaderColumnNameMappingStrategy会只映射CSV表头中与StatusObject字段匹配的列，其余列直接跳过，无需额外处理。
• 分隔符与空格处理：通过setColumnSeparator('|')适配你的CSV格式，setTrim(true)自动清理字段前后的空格（比如示例中12 | abc的空格会被去除）。
• 类型匹配：确保StatusObject的字段类型与CSV中的数据类型一致，例如CSV中的age是整数，字段需用Integer而非String。

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