I want to perform aggregation processing with spring-batch

Background

In a WEB application made with Spring-Boot I want to implement such batch processing that aggregates the contents of the prepared API execution log file and stores the result in the DB. However, there is no site that can be used as a reference for implementing that requirement, and it is ridiculously clogged, so I will leave it as a memorandum. (I don't know if this is the best)

• About 7 days of batch history at the time of posting by us

Main requirements

• I want to aggregate for each specific item in the data
• Aggregation is done on the Java side (not in SQL)
• Convert a part of data to a character string mapped before storing in DB

Prerequisite data (example)

Before: Log format to be aggregated

ʻURL / HTTP method / HTTP status code / execution time / execution date and timeat the time of access are listed in TSV format. Example: / api / ・ ・ ・ GET 200 48 2020/08/14 11:05:42 701 / api / ・ ・ ・ GET 200 27 2020/08/14 11:05:43 352 / api / ・ ・ ・ ・ / 41 DELETE 401 10 2020/08/14 11:05:46 780 / api / ・ ・ ・ / 42 PUT 200 108 2020/08/14 11:06:16 824 / api / ・ ・ ・ POST 500 806 2020/08/14 11:06:30 252 ・ ・ ・ `

After: Format when storing DB

• ʻURL` is processed into a logical API name
• ʻURL (API name) / HTTP method / HTTP status code is used as a key to aggregate access countandaverage execution time`.
• Record format: ʻAPI name / HTTP method / HTTP status code / number of accesses / average execution time / total date and time`
• Example: ○○ API GET 200 10 240

Implementation

policy

• Implemented with chunk model
• Reader: Read line by line (1 line of data = 1Item)
• Processor: Processing of Item sent from Reader
• Writer: Aggregate the list of processed items sent from Processor → Prepare another Writer instance and pass the aggregated result to it

point

• Perform aggregation processing with Stream API
• Prepare two, a Writer that receives the Item group from the Processor and a Writer that actually performs the writing process.

code

• Processing around batch settings and contents of Entity class are omitted.

Dto class

@Data
public class LogCollectedDto {
  //API name
  private String apiName;
  //HTTP method
  private String httpMethod;
  //HTTP status code
  private String httpCode;
  //Execution time(ms)
  private String executionTime;
  //Aggregation date and time
  private String collectedDate;
}

Reader

Defined in Bean

  @Bean
  public FlatFileItemReader<LogCollectedDto> reader() {

    final String READ_FILE_PATH = <Log file name to read>;

    FlatFileItemReader<LogCollectedDto> reader = new FlatFileItemReader<>();
    reader.setResource(new FileSystemResource(READ_FILE_PATH));
    reader.setEncoding(StandardCharsets.UTF_8.name());
    reader.setLinesToSkip(0);
    reader.setLineMapper(
        new DefaultLineMapper() {
          {
            setLineTokenizer(
                new DelimitedLineTokenizer(DelimitedLineTokenizer.DELIMITER_TAB) {
                  {
                    setNames(
                        new String[] {
                          "apiUrl", "httpMethod", "httpCode", "executionTime", "collectedDate"
                        });
                  }
                });
            setFieldSetMapper(
                new BeanWrapperFieldSetMapper<LogCollectedDto>() {
                  {
                    setTargetType(LogCollectedDto.class);
                  }
                });
          }
        });
    return reader;
  }

Processor

Cut out to another class

public class CustomItemProcessor implements ItemProcessor<LogCollectedDto, LogCollectedDto> {

  @Override
  public LogCollectedDto process(LogCollectedDto item) throws Exception {

    //Validation of the retrieved Item, skip that line if false
    if (!validateItem(item)) {
      return null;
    }

    //Hold the acquired Item once in another variable for processing later
    //(If you modify the argument directly, the Item acquired when interrupting and restarting may become the processed data)
    LogCollectedDto afterItem = item;
    //Data content processing (separate method omitted)
    afterItem.setApiName(getApiName(・ ・ ・));

    return afterItem;
  }

//(Omitted)

}

Writer

Cut out to another class

@RequiredArgsConstructor
public class CustomItemWriter extends JpaItemWriter<LogCollectedDto> {

  private final JpaItemWriter<Log> jpaItemWriter;

  @Override
  public void write(List<? extends LogCollectedDto> items) {

   //Aggregate the Items received from the Processor and pass it to another Writer instance
    Map<String, List<LogCollectedDto>> groupingMap = groupingItems(items);
    jpaItemWriter.write(collectItems(groupingMap));
  }

  /**
   *Grouping Items received from Processor
   *API name and HTTP status as composite key
   *
   * @param list
   * @return
   */
  private Map<String, List<LogCollectedDto>> groupingItems(List<? extends LogCollectedDto> list) {
    //Create a composite key
    Function<LogCollectedDto, String> compositeKey =
        logDto -> {
          StringBuffer sb = new StringBuffer();
          sb.append(logDto.getApiName()).append("-").append(logDto.getHttpMethod());
          return sb.toString();
        };

    Map<String, List<LogCollectedDto>> grpByComplexKeys =
        list.stream().collect(Collectors.groupingBy(compositeKey));

    return grpByComplexKeys;
  }

  /**
   *Generate a list of Entity by aggregating grouped items
   *
   * @param groupingMap
   * @return
   */
  private List<Log> collectItems(Map<String, List<LogCollectedDto>> groupingMap) {

    List<Log> recordList = new ArrayList<>();

    for (List<LogCollectedDto> dtoList : groupingMap.values()) {
      //Instantiation of Entity class
      Log record = new Log();
      //Aggregation processing
      record.setApiName(dtoList.stream().findFirst().get().getApiName());
      record.setHttpCode(dtoList.stream().findFirst().get().getHttpCode());
      record.setHttpMethod(dtoList.stream().findFirst().get().getHttpMethod());
      record.setAccesses(dtoList.size());
      record.setAverageTime(
          dtoList.stream()
              .collect(
                  Collectors.averagingDouble(dto -> Double.parseDouble(dto.getExecutionTime()))));
      record.setCollectedTime(LocalDateTime.now());
      recordList.add(record);
    }

    return recordList;
  }
}

Concern

• With only the above implementation, the number of rows that can be aggregated at one time will be up to the set chunk size.
• → It seems that the above should be taken as one step, and if the same key exists in the contents stored in the DB in the next step, it should be implemented to recount (it seems that the DB load is high ...)
• I cannot deny the feeling that the writer's responsibilities are cluttered.

Summary

There are concerns, but I personally think it's the cleanest way to write. We are always looking for opinions. Conclusion: Stream API is the strongest! !! !!