数据库的春季批量分区无法正常工作

Question

我已经配置了一个工作如下，它是从数据库读取并写入文件，但通过按顺序对数据进行分区。

//作业配置

@Bean
    public Job job(JobBuilderFactory jobBuilderFactory) throws Exception {
        Flow masterFlow1 = (Flow) new FlowBuilder<Object>("masterFlow1").start(masterStep()).build();
        return (jobBuilderFactory.get("Partition-Job")
                .incrementer(new RunIdIncrementer())
                .start(masterFlow1)
                .build()).build();
    }


    @Bean
    public Step masterStep() throws Exception
    {
        return stepBuilderFactory.get(MASTERPPREPAREDATA)
                //.listener(customSEL)
                .partitioner(STEPPREPAREDATA,new  DBPartitioner())
                .step(prepareDataForS1())
                .gridSize(gridSize)
                .taskExecutor(new SimpleAsyncTaskExecutor("Thread"))
                .build();
    }

    @Bean
    public Step prepareDataForS1() throws Exception
    {
        return stepBuilderFactory.get(STEPPREPAREDATA)
                //.listener(customSEL)
                .<InputData,InputData>chunk(chunkSize)
                .reader(JDBCItemReader(0,0))
                .writer(writer(null))
                .build();
    }

@Bean(destroyMethod="")
    @StepScope
    public JdbcCursorItemReader<InputData> JDBCItemReader(@Value("#{stepExecutionContext[startingIndex]}") int startingIndex,
            @Value("#{stepExecutionContext[endingIndex]}") int endingIndex)
    {
        JdbcCursorItemReader<InputData> ir = new JdbcCursorItemReader<>();
        ir.setDataSource(batchDataSource);
        ir.setMaxItemCount(DBPartitioner.partitionSize);
        ir.setSaveState(false);
        ir.setRowMapper(new InputDataRowMapper());
        ir.setSql("SELECT * FROM FIF_INPUT fi WHERE fi.SEQ > ? AND fi.SEQ < ?");
        ir.setPreparedStatementSetter(new PreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps) throws SQLException {
                ps.setInt(1, startingIndex);
                ps.setInt(2, endingIndex);
            }
        });
        return ir;
    }

@Bean
    @StepScope
    public FlatFileItemWriter<InputData> writer(@Value("#{stepExecutionContext[index]}") String index)
    {
        System.out.println("writer initialized!!!!!!!!!!!!!"+index);
        //Create writer instance
        FlatFileItemWriter<InputData> writer = new FlatFileItemWriter<>();

        //Set output file location
        writer.setResource(new FileSystemResource(batchDirectory+relativeInputDirectory+index+inputFileForS1));

        //All job repetitions should "append" to same output file
        writer.setAppendAllowed(false);


        //Name field values sequence based on object properties
        writer.setLineAggregator(customLineAggregator);
        return writer;
    }

为db分区提供的Partitioner单独写在其他文件中，如下

//PartitionDb.java

public class DBPartitioner implements Partitioner{



    public static int partitionSize;
    private static Log log = LogFactory.getLog(DBPartitioner.class);
    @SuppressWarnings("unchecked")
    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {

        log.debug("START: Partition"+"grid size:"+gridSize);

        @SuppressWarnings("rawtypes")
        Map partitionMap = new HashMap<>();
        int startingIndex = -1;
        int endSize = partitionSize+1;


        for(int i=0; i< gridSize; i++){
            ExecutionContext ctxMap = new ExecutionContext();
            ctxMap.putInt("startingIndex",startingIndex);
            ctxMap.putInt("endingIndex", endSize);
            ctxMap.put("index", i);
            startingIndex = endSize-1;
            endSize += partitionSize; 
            partitionMap.put("Thread:-"+i, ctxMap);
        }
        log.debug("END: Created Partitions of size: "+ partitionMap.size());
        return partitionMap;
    }




}

这个执行正确，但问题是即使在基于序列进行分区之后，我在多个文件中得到相同的行，这是不正确的，因为我为每个分区提供了不同的数据集。谁能告诉我怎么了。我正在使用 HikariCP 进行 Db 连接池和 Spring Batch 4

Answer 1

这个执行正确，但问题是即使在根据序列进行分区之后，我在多个文件中得到相同的行，这是不正确的，因为我为每个分区提供不同的数据集。

我不确定您的分区程序是否正常工作。快速测试表明它没有提供您声称的不同数据集：

DBPartitioner dbPartitioner = new DBPartitioner();
Map<String, ExecutionContext> partition = dbPartitioner.partition(5);
for (String s : partition.keySet()) {
    System.out.println(s + " : " + partition.get(s));
}

打印出来：

Thread:-0 : {endingIndex=1, index=0, startingIndex=-1}
Thread:-1 : {endingIndex=1, index=1, startingIndex=0}
Thread:-2 : {endingIndex=1, index=2, startingIndex=0}
Thread:-3 : {endingIndex=1, index=3, startingIndex=0}
Thread:-4 : {endingIndex=1, index=4, startingIndex=0}

如您所见，几乎所有分区都有相同的startingIndex 和endingIndex。

我建议您在分区步骤中使用分区器之前对其进行单元测试。