[JAVA] Je souhaite effectuer un traitement d'agrégation avec spring-batch

Contexte

Dans une application WEB réalisée avec Spring-Boot Je souhaite implémenter un tel processus par lots qui agrège le contenu du fichier journal d'exécution de l'API préparé et stocke le résultat dans la base de données. Cependant, aucun site ne pourrait être utilisé comme référence pour la mise en œuvre de cette exigence, et il était ridiculement bouché, je vais donc le laisser comme un mémorandum. (Je ne sais pas si c'est le meilleur)

• Environ 7 jours d'historique des lots au moment de la publication par nous

Principales exigences

• Je souhaite agréger pour chaque élément spécifique dans les données
• L'agrégation se fait côté Java (pas en SQL)
• Convertir une partie des données en une chaîne de caractères mappée avant de les stocker dans DB

Données prérequises (exemple)

Avant: format de journal à agréger

ʻURL / méthode HTTP / code d'état HTTP / heure d'exécution / date et heure d'exécutionau moment de l'accès sont répertoriés au format TSV Exemple: / api / ・ ・ ・ GET 200 48 14/08/2020 11:05:42 701 / api / ・ ・ ・ GET 200 27 2020/08/14 11:05:43 352 / api / ・ ・ ・ / 41 SUPPRIMER 401 10 2020/08/14 11:05:46 780 / api / ・ ・ ・ / 42 PUT 200108 2020/08/14 11:06:16 824 / api / ・ ・ ・ POST 500806 2020/08/14 11:06:30 252 ・ ・ ・ `

Après: Format lors du stockage de DB

• ʻURL` est transformé en un nom d'API logique
• ʻEn utilisant l'URL (nom de l'API) / la méthode HTTP / le code d'état HTTPcomme clé, lenombre d'accès et le temps d'exécution moyen` sont agrégés.
• Format d'enregistrement: ʻAPI name / HTTP method / HTTP status code / number of accesses / average execution time / total date and time`
• Exemple: ○○ API GET 200 10240

la mise en oeuvre

politique

• Implémenté avec chunk model
• Lecteur: lecture ligne par ligne (1 ligne de données = 1 article)
• Processeur: Traitement des éléments envoyés par Reader
• Writer: agréger la liste des éléments traités envoyés par le processeur → préparer une autre instance de Writer et lui transmettre le résultat agrégé

point

• Effectuer le traitement d'agrégation avec Stream API
• Préparez-en deux, un rédacteur qui reçoit le groupe d'éléments du processeur et un rédacteur qui effectue réellement le processus d'écriture.

code

• Le traitement autour des paramètres de lot et du contenu de la classe Entity est omis.

Classe Dto

@Data
public class LogCollectedDto {
  //Nom de l'API
  private String apiName;
  //Méthode HTTP
  private String httpMethod;
  //Code d'état HTTP
  private String httpCode;
  //Temps d'exécution(ms)
  private String executionTime;
  //Date et heure d'agrégation
  private String collectedDate;
}

Reader

Défini dans Bean

  @Bean
  public FlatFileItemReader<LogCollectedDto> reader() {

    final String READ_FILE_PATH = <Nom du fichier journal à lire>;

    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

Couper à une autre classe

public class CustomItemProcessor implements ItemProcessor<LogCollectedDto, LogCollectedDto> {

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

    //Validation de l'élément récupéré, ignorez cette ligne si false
    if (!validateItem(item)) {
      return null;
    }

    //Conserver l'élément acquis une fois dans une autre variable pour un traitement ultérieur
    //(Si vous modifiez directement l'argument, l'élément acquis lors de l'interruption et du redémarrage peut devenir les données traitées)
    LogCollectedDto afterItem = item;
    //Traitement du contenu des données (méthode distincte omise)
    afterItem.setApiName(getApiName(・ ・ ・));

    return afterItem;
  }

//(Omis)

}

Writer

Couper à une autre classe

@RequiredArgsConstructor
public class CustomItemWriter extends JpaItemWriter<LogCollectedDto> {

  private final JpaItemWriter<Log> jpaItemWriter;

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

   //Regroupez les éléments reçus du processeur et transmettez-les à une autre instance de Writer
    Map<String, List<LogCollectedDto>> groupingMap = groupingItems(items);
    jpaItemWriter.write(collectItems(groupingMap));
  }

  /**
   *Regroupement des éléments reçus du processeur
   *Nom de l'API et statut HTTP en tant que clé composite
   *
   * @param list
   * @return
   */
  private Map<String, List<LogCollectedDto>> groupingItems(List<? extends LogCollectedDto> list) {
    //Créer une clé composite
    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;
  }

  /**
   *Générer une liste d'entités en agrégeant des éléments groupés
   *
   * @param groupingMap
   * @return
   */
  private List<Log> collectItems(Map<String, List<LogCollectedDto>> groupingMap) {

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

    for (List<LogCollectedDto> dtoList : groupingMap.values()) {
      //Instanciation de classe d'entité
      Log record = new Log();
      //Traitement d'agrégation
      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;
  }
}

Préoccupation

• Avec uniquement l'implémentation ci-dessus, le nombre de lignes pouvant être agrégées en une seule fois sera jusqu'à la taille de bloc définie.
• → Prenez ce qui précède comme une étape, et si la même clé existe dans le contenu stocké dans la base de données à l'étape suivante, il semble qu'elle devrait être implémentée pour recomptage (la charge de la base de données semble être élevée ...)
• Je ne peux nier le sentiment que les responsabilités de l'écrivain sont encombrées.

Résumé

Il y a des inquiétudes, mais je pense personnellement que c'est la façon la plus propre d'écrire. Nous recherchons toujours des opinions. Conclusion: l'API Stream est la plus puissante! !! !!