Spring Batch导出CSV出现重复/缺失数据的原因及解决方案

使用MongoItemReader生成CSV文件时，发现输出文件存在重复记录和缺失记录的问题，即使已将chunk size设置为1。以下是相关配置代码及版本信息，希望了解问题原因及解决方法。

相关配置代码 #

1. BatchApplication.java配置 #

@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
public class BatchApplication {
    public static void main(String[] args) {
        SpringApplication.run(BatchApplication.class, args);
    }
}

2. Job与Step配置类BatchConfig #

@Configuration
@EnableBatchProcessing
@Slf4j
public class BatchConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    @Qualifier(value = "testMongoTemplate")
    private MongoTemplate MongoTemplate;
    @Autowired
    private BatchProperties batchProperties;
    @Autowired
    private MyJobListener myJobListener;
    @Autowired
    private BatchHelperService batchHelperService;
    @Autowired
    private RecordProcessor recordProcessor;
    @Autowired
    private S3Client s3Client;
    @Autowired
    private S3Processor s3Processor;
    @Autowired
    private ReportFileWriter reportFileWriter;


    @StepScope
    public MongoItemReader<AppModel> readerOffering(final long startDateTimeEpoch, final long endDateTimeEpoch) {

        final MongoItemReader<InsuranceAppModel> reader = new MongoItemReader<>();
        reader.setTemplate(mongoTemplate);
        reader.setSort(Map.of("lastUpdatedDateTime", Sort.Direction.ASC));
        reader.setTargetType(AppModel.class);
        reader.setCollection("my_table");
        reader.setQuery("{'lastUpdatedDateTime' : { '$gte' : " + startDateTimeEpoch + ", '$lte' : " + endDateTimeEpoch + "}}");
        return reader;
    }

    @Bean
    public Job applicationReportJob() throws Exception {

        final List<Step> stepList = new ArrayList<>();


        stepList.add(stepBuilderFactory.get("step_1_query_db").<AppModel, CSVReport>chunk(1)
                .reader(readerOffering(1698858000000L, 1698944399000L))
                .processor(recordProcessor)
                .writer(reportFileWriter)
                .build());
        stepList.add(stepBuilderFactory.get("step_2_upload_to_S3").tasklet(s3Processor).build());


        final SimpleJobBuilder jobBuilder = jobBuilderFactory.get("applicationReportJob").listener(myjobListener).incrementer(new RunIdIncrementer()).start(stepList.get(0));
        stepList.remove(0);
        for (final Step step : stepList) {
            jobBuilder.next(step);
        }
        return jobBuilder.build();
    }
}

3. RecordProcessor处理器代码 #

@Component
@Slf4j
public class RecordProcessor implements ItemProcessor<AppModel, CSVReport> {

    @Autowired
    private BatchProperties batchProperties;

    @Override
    public CSVReport process(final AppModel record) throws Exception {
        CSVReport.CSVReportBuilder recordCsv = CSVReport.builder();
        try {
            recordCsv.applicationId(record.getApplicationId());
            recordCsv.status(record.getSstatus());
        } catch (Exception ex) {
            throw new BatchException(ex.getMessage(), ex, "ERROR_WHILE_PROCESSING__DATA");
        }
        return record.build();
    }

}

4. ReportFileWriter写入器代码 #

@Component
@Slf4j
public class ReportFileWriter implements ItemWriter<CSVReport> {
    @Autowired
    private BatchProperties batchProperties;
    @Autowired
    private BatchHelperService batchHelperService;
    
    @Override
    public void write(List<? extends CSVReport> items) throws Exception {
        StringBuilder sb = new StringBuilder();
        
        for (CSVReport item : items) {
            try {
                sb.append("\n");
                sb.append(item.getApplicationId());
                sb.append("|");
                sb.append(item.getStatus());

            } catch (Exception ex) {
                log.error("Error while append string");
            }
        }

        if(StringUtils.isNotBlank(record)) {
            try (OutputStreamWriter fileWriter = new OutputStreamWriter(new FileOutputStream("output/" +  batchHelperService.getBatchProcessingModel().getFileName(), true), StandardCharsets.UTF_8)) {
                fileWriter.write(sb.toString().replace("null", ""));
            } catch (Exception ex) {
                throw new BatchException(ex.getMessage(), ex, "ERROR_WHILE_WRITING_FILE");
            }
        }

    }

}

Spring Batch版本说明 #

Spring Batch版本为截图中所示版本。

问题原因分析 #

1. 排序字段不唯一：当前用lastUpdatedDateTime作为排序字段，若多条记录该字段值相同，MongoItemReader的分页查询会出现重复或遗漏——因为分页依赖排序字段定位，重复值会导致分页边界的记录被重复读取或跳过。
2. Writer追加模式问题：自定义Writer使用FileOutputStream追加模式（第二个参数为true），若Job因失败重启，会将已处理的记录再次追加到文件，造成重复。
3. 处理器错误处理缺陷：处理器捕获异常后抛出BatchException，但未配置跳过策略，会触发Chunk重试，导致同一条记录被重复处理；同时代码中record.getSstatus()疑似拼写错误（应为record.getStatus()），会导致字段值为null，在Writer中被替换为空，表现为记录缺失。
4. StepScope未正确生效：Reader方法标注了@StepScope，但直接在Job定义中传入参数，导致Reader未被Spring容器代理，出现状态不一致问题。

解决方法 #

1. 确保排序字段唯一 #

修改MongoItemReader的排序配置，加入唯一字段（如_id）保证分页准确性：

reader.setSort(Map.of("lastUpdatedDateTime", Sort.Direction.ASC, "_id", Sort.Direction.ASC));

2. 修复Writer写入逻辑 #

• 取消追加模式，每次Job执行创建新文件，避免重复写入：

  try (OutputStreamWriter fileWriter = new OutputStreamWriter(new FileOutputStream("output/" + batchHelperService.getBatchProcessingModel().getFileName()), StandardCharsets.UTF_8)) {
      fileWriter.write(sb.toString().replace("null", ""));
  }
  

• 或在Job启动时（通过JobListener）清空目标文件。

3. 优化处理器与错误处理 #

• 修正拼写错误：将record.getSstatus()改为record.getStatus()，避免字段值丢失。
• 配置跳过策略，允许跳过处理失败的记录，防止Chunk重试导致重复：

  stepBuilderFactory.get("step_1_query_db").<AppModel, CSVReport>chunk(1)
          .reader(readerOffering(1698858000000L, 1698944399000L))
          .processor(recordProcessor)
          .writer(reportFileWriter)
          .faultTolerant()
          .skip(BatchException.class)
          .skipLimit(10) // 根据业务需求设置跳过次数
          .build();
  

4. 正确使用StepScope #

通过Job参数传递Reader的时间范围，确保StepScope代理生效：

@StepScope
public MongoItemReader<AppModel> readerOffering(
        @Value("#{jobParameters['startDateTimeEpoch']}") long startDateTimeEpoch,
        @Value("#{jobParameters['endDateTimeEpoch']}") long endDateTimeEpoch) {
    // 原Reader逻辑不变
}

在Job启动时传入参数，或在Step定义中使用参数占位符。

5. 使用内置FlatFileItemWriter替代自定义Writer #

Spring Batch内置的FlatFileItemWriter支持事务性写入，能避免重复和缺失问题，且无需手动处理文件流：

@StepScope
@Bean
public FlatFileItemWriter<CSVReport> csvReportWriter() {
    return new FlatFileItemWriterBuilder<CSVReport>()
            .name("csvReportWriter")
            .resource(new FileSystemResource("output/" + batchHelperService.getBatchProcessingModel().getFileName()))
            .delimited()
            .delimiter("|")
            .names("applicationId", "status")
            .build();
}

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