Весеннее пакетное разбиение базы данных не работает должным образом - PullRequest
0 голосов
/ 06 февраля 2019

Я настроил задание следующим образом: чтение из базы данных и запись в файлы, но путем разделения данных на основе последовательности.

// Конфигурация задания

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

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




}

Этот файл выполняется правильно, но проблема возникает даже после разбиения на основе последовательности iя получаю одинаковые строки в нескольких файлах, что неправильно, поскольку я предоставляю различный набор данных для каждого раздела.Может кто-нибудь сказать мне, что не так.Я использую HikariCP для пулов соединений Db и пакетной пружины 4

1 Ответ

0 голосов
/ 07 февраля 2019

Этот выполняется правильно, но проблема даже после разбиения на основе последовательности. Я получаю одинаковые строки в нескольких файлах, что неправильно, поскольку я предоставляю разный набор данных для каждого раздела.

Я не уверен, что ваш разделитель работает правильно.Быстрый тест показывает, что он не предоставляет различные наборы данных, как вы утверждаете:

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.

Я рекомендую вам выполнить модульное тестирование вашего разделителя, прежде чем использовать его на шаге с разбивкой.

...