DB的Spring Batch的分区不能正常工作

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;
    }

提供了一种用于划分分贝分割器被如此如下在其它文件分开写入

//partition dB.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的数据库连接池，春季批次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。

我建议你单位分区步骤中使用它之前测试你的分区。