1
\$\begingroup\$

We are using spring-batch to extract data from a SAP system, during processing we pass the extracted full texts to a tagging service, in the end the enriched documents are stored in Solr to be served to the user.

The problem during this ETL chain is that the tagging service is slow. Unfortunately we cannot improve its performance problems as it is a bought vendor solution. The problem of this service is that it has a ramp-up time of 6 seconds for each request. It does offer a bulk interface to tackle this. So we can send 1 item for tagging or 5.000 items. Each request will take 6 seconds. Hence we would like to send the texts as bulk during processing.

This is where our review request comes into play.

As far as we understood spring-batch processors perform their operation on single records, not batches. Hence we have a problem, as we need to process 8 Million records.

I have setup a project on github to illustrate the problem and our approach so far. To start the batch processing use the SlowProcessorApplicationTests.

The reader is a FlatFileItemReader as the sample data is included as CSV within the project

@Bean
public FlatFileItemReader<InputItem> inputItemReader() {
  return new FlatFileItemReaderBuilder<InputItem>()
      .name("inputItemReader")
      .resource(new ClassPathResource("input.csv"))
      .linesToSkip(1)
      .delimited()
      .delimiter(";")
      .names(new String[]{
          "id",
          "title",
          "text"})
      .fieldSetMapper(new BeanWrapperFieldSetMapper<InputItem>() {{
        setTargetType(InputItem.class);
      }})
      .build();
}

The ItemProcessor is straight forward, but here is where the problem is. The call to taggingService.analyze(...) takes 6 seconds, each time.

@Slf4j
@Component
public class CSVItemProcessor implements ItemProcessor<InputItem, OutputItem> {

  @Autowired
  private TaggingService taggingService;

  @Override
  public OutputItem process(InputItem inItem) throws InterruptedException {
    log.trace("process {}", inItem);
    OutputItem outItem = new OutputItem();
    outItem.setId(inItem.getId());
    outItem.setTitle(inItem.getTitle());

    List<TaggingResponse> responses = taggingService.analyze(inItem.getText());
    TaggingResponse response = responses.get(0);
    outItem.setTags(response.getTags());
    outItem.setWords(response.getWords());
    return outItem;
  }
}

The Job and its only Step are also quite simple:

@Bean
public Job loadCSV(Step handleCSV) {

  return jobBuilder.get("loadCSV")
      .incrementer(new RunIdIncrementer())
      .start(handleCSV)
      .build();
}

@Bean
public Step handleCSV(
    FlatFileItemReader<InputItem> inputItemReader,
    CSVItemProcessor itemProcessor) {

  return stepFactory.get("handleCSV")
    .<InputItem, OutputItem>chunk(2)
    .reader(inputItemReader)
    .processor(itemProcessor)
    .writer((ItemWriter<Object>) items -> log.debug("writing {} items", items.size()))
    .build();
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.