2
\$\begingroup\$

I have a simple .csv file with ~400,000 line (one column only). It takes me a lot of time to read the records and process them.

The processor validating records against couchbase. The writer - writing into remote topic, takes me around 30 mins.

I read that flatfileItemreader is not thread-safe, so my chunk value is 1. I read the asynchronous processing could assist, but I can't see any improvements.

@Configuration
@EnableBatchProcessing
public class NotificationFileProcessUploadedFileJob {


    @Value("${expected.snid.header}")
    public String snidHeader;

    @Value("${num.of.processing.chunks.per.file}")
    public int numOfProcessingChunksPerFile;

    @Autowired
    private InfrastructureConfigurationConfig infrastructureConfigurationConfig;

    private static final String OVERRIDDEN_BY_EXPRESSION = null;


    @Inject
    private JobBuilderFactory jobs;

    @Inject
    private StepBuilderFactory stepBuilderFactory;

    @Inject
    ExecutionContextPromotionListener executionContextPromotionListener;


    @Bean
    public Job processUploadedFileJob() throws Exception {
        return this.jobs.get("processUploadedFileJob").start((processSnidUploadedFileStep())).build();

    }

    @Bean
    public Step processSnidUploadedFileStep() {
        return stepBuilderFactory.get("processSnidFileStep")
                .<PushItemDTO, PushItemDTO>chunk(numOfProcessingChunksPerFile)
                .reader(snidFileReader(OVERRIDDEN_BY_EXPRESSION))
                .processor(asyncItemProcessor())
                .writer(asyncItemWriter())
            //    .throttleLimit(20)
             //   .taskJobExecutor(infrastructureConfigurationConfig.taskJobExecutor())


                        //     .faultTolerant()
                        //   .skipLimit(10) //default is set to 0
                        //     .skip(MySQLIntegrityConstraintViolationException.class)
                .build();
    }

    @Inject
    ItemWriter writer;

    @Bean
    public AsyncItemWriter asyncItemWriter() {
        AsyncItemWriter asyncItemWriter=new AsyncItemWriter();
        asyncItemWriter.setDelegate(writer);
        return asyncItemWriter;
    }


    @Bean
    @Scope(value = "step", proxyMode = ScopedProxyMode.INTERFACES)
    public ItemStreamReader<PushItemDTO> snidFileReader(@Value("#{jobParameters[filePath]}") String filePath) {
        FlatFileItemReader<PushItemDTO> itemReader = new FlatFileItemReader<PushItemDTO>();
        itemReader.setLineMapper(snidLineMapper());
        itemReader.setLinesToSkip(1);
        itemReader.setResource(new FileSystemResource(filePath));
        return itemReader;
    }


    @Bean
    public AsyncItemProcessor asyncItemProcessor() {

        AsyncItemProcessor<PushItemDTO, PushItemDTO> asyncItemProcessor = new AsyncItemProcessor();

        asyncItemProcessor.setDelegate(processor(OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION,
                OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION));
        asyncItemProcessor.setTaskExecutor(infrastructureConfigurationConfig.taskProcessingExecutor());

        return asyncItemProcessor;

    }

    @Scope(value = "step", proxyMode = ScopedProxyMode.INTERFACES)
    @Bean
    public ItemProcessor<PushItemDTO, PushItemDTO> processor(@Value("#{jobParameters[pushMessage]}") String pushMessage,
                                                             @Value("#{jobParameters[jobId]}") String jobId,
                                                             @Value("#{jobParameters[taskId]}") String taskId,
                                                             @Value("#{jobParameters[refId]}") String refId,
                                                             @Value("#{jobParameters[url]}") String url,
                                                             @Value("#{jobParameters[targetType]}") String targetType,
                                                             @Value("#{jobParameters[gameType]}") String gameType) {
        return new PushItemProcessor(pushMessage, jobId, taskId, refId, url, targetType, gameType);
    }

    @Bean
    public LineMapper<PushItemDTO> snidLineMapper() {
        DefaultLineMapper<PushItemDTO> lineMapper = new DefaultLineMapper<PushItemDTO>();
        DelimitedLineTokenizer lineTokenizer = new DelimitedLineTokenizer();
        lineTokenizer.setDelimiter(",");
        lineTokenizer.setStrict(true);
        lineTokenizer.setStrict(true);
        String[] splittedHeader = snidHeader.split(",");
        lineTokenizer.setNames(splittedHeader);
        BeanWrapperFieldSetMapper<PushItemDTO> fieldSetMapper = new BeanWrapperFieldSetMapper<PushItemDTO>();
        fieldSetMapper.setTargetType(PushItemDTO.class);

        lineMapper.setLineTokenizer(lineTokenizer);
        lineMapper.setFieldSetMapper(new PushItemFieldSetMapper());
        return lineMapper;
    }
}


 @Bean
    @Override
    public SimpleAsyncTaskExecutor taskProcessingExecutor() {
        SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
        simpleAsyncTaskExecutor.setConcurrencyLimit(300);
        return simpleAsyncTaskExecutor;
    }

How do you think I could improve the processing performances and make them faster?

\$\endgroup\$
1
  • \$\begingroup\$ Javadoc of SimpleAsyncTaskExecutor has a big warning that recommends a thread-pooling TaskExecutor. \$\endgroup\$
    – Pino
    Sep 7, 2017 at 7:24

2 Answers 2

1
\$\begingroup\$

commit-interval in 100-1000 would help you for sure. I implemented task executor in a similar case. But since a batch reader was too dependent on a number of pre-conditions, multiple requests never hampered the result data and performance improvement was manifold. Maybe you can test your corner cases and try testing code with executor with appropriate thread pool size.

\$\endgroup\$
1
\$\begingroup\$

A few things to note here:

I don't think the actual slowness comes from reading the file, because this flatfile of only one column should be read as fast as you can write pretty much, I don't think it's the bottleneck in your case. It's much more likely that writing to your datasource is causing the slowdown. Can you verify that? I also of commit-interval in your config, a high number will make it much much faster. If you are using a commit interval of something like 1, it will be painfully slow. Please refer to this thread.

If you are absolutely positive that reading the file is the bottleneck, you can always implement the first step as "partitioner" and then instead using the FlatFileItemReader, use one of the partitioning readers in the next step. This will be thread safe.

\$\endgroup\$
1
  • \$\begingroup\$ ill need to split my big input file into small ones before right? \$\endgroup\$
    – rayman
    Feb 21, 2015 at 8:54

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.