hudi-client/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java
/**
   * Performs cleaning of partition paths according to cleaning policy and returns the number of files cleaned. Handles
   * skews in partitions to clean by making files to clean as the unit of task distribution.
   *
   * @throws IllegalArgumentException if unknown cleaning policy is provided
   */
Performs cleaning of partition paths according to cleaning policy and returns the number of files cleaned. Handles
skews in partitions to clean by making files to clean as the unit of task distribution.

@throws IllegalArgumentException if unknown cleaning policy is provided
Performs cleaning of partition paths according to cleaning policy and returns the number of files cleaned . Handles skews in partitions to clean by making files to clean as the unit of task distribution . @throws IllegalArgumentException if unknown cleaning policy is provided
List<HoodieCleanStat> clean(JavaSparkContext jsc, HoodieCleanerPlan cleanerPlan) {
    int cleanerParallelism = Math.min(
        (int) (cleanerPlan.getFilePathsToBeDeletedPerPartition().values().stream().mapToInt(List::size).count()),
        config.getCleanerParallelism());
    LOG.info("Using cleanerParallelism: " + cleanerParallelism);
    jsc.setJobGroup(this.getClass().getSimpleName(), "Perform cleaning of partitions");
    List<Tuple2<String, PartitionCleanStat>> partitionCleanStats = jsc
        .parallelize(cleanerPlan.getFilePathsToBeDeletedPerPartition().entrySet().stream()
            .flatMap(x -> x.getValue().stream().map(y -> new Tuple2<>(x.getKey(),
              new CleanFileInfo(y.getFilePath(), y.getIsBootstrapBaseFile()))))
            .collect(Collectors.toList()), cleanerParallelism)
        .mapPartitionsToPair(deleteFilesFunc(table))
        .reduceByKey(PartitionCleanStat::merge).collect();
    Map<String, PartitionCleanStat> partitionCleanStatsMap = partitionCleanStats.stream()
        .collect(Collectors.toMap(Tuple2::_1, Tuple2::_2));
    return cleanerPlan.getFilePathsToBeDeletedPerPartition().keySet().stream().map(partitionPath -> {
      PartitionCleanStat partitionCleanStat = partitionCleanStatsMap.containsKey(partitionPath)
          ? partitionCleanStatsMap.get(partitionPath)
          : new PartitionCleanStat(partitionPath);
      HoodieActionInstant actionInstant = cleanerPlan.getEarliestInstantToRetain();
      return HoodieCleanStat.newBuilder().withPolicy(config.getCleanerPolicy()).withPartitionPath(partitionPath)
          .withEarliestCommitRetained(Option.ofNullable(
              actionInstant != null
                  ? new HoodieInstant(HoodieInstant.State.valueOf(actionInstant.getState()),
                  actionInstant.getAction(), actionInstant.getTimestamp())
                  : null))
          .withDeletePathPattern(partitionCleanStat.deletePathPatterns())
          .withSuccessfulDeletes(partitionCleanStat.successDeleteFiles())
          .withFailedDeletes(partitionCleanStat.failedDeleteFiles())
          .withDeleteBootstrapBasePathPatterns(partitionCleanStat.getDeleteBootstrapBasePathPatterns())
          .withSuccessfulDeleteBootstrapBaseFiles(partitionCleanStat.getSuccessfulDeleteBootstrapBaseFiles())
          .withFailedDeleteBootstrapBaseFiles(partitionCleanStat.getFailedDeleteBootstrapBaseFiles())
          .build();
    }).collect(Collectors.toList());
  }
List < HoodieCleanStat > clean ( JavaSparkContext jsc , HoodieCleanerPlan cleanerPlan ) { int cleanerParallelism = Math . min ( ( int ) ( cleanerPlan . getFilePathsToBeDeletedPerPartition ( ) . values ( ) . stream ( ) . mapToInt ( List :: size ) . count ( ) ) , config . getCleanerParallelism ( ) ) ; LOG . info ( "Using cleanerParallelism: " + cleanerParallelism ) ; jsc . setJobGroup ( this . getClass ( ) . getSimpleName ( ) , "Perform cleaning of partitions" ) ; List < Tuple2 < String , PartitionCleanStat > > partitionCleanStats = jsc . parallelize ( cleanerPlan . getFilePathsToBeDeletedPerPartition ( ) . entrySet ( ) . stream ( ) . flatMap ( x -> x . getValue ( ) . stream ( ) . map ( y -> new Tuple2 < > ( x . getKey ( ) , new CleanFileInfo ( y . getFilePath ( ) , y . getIsBootstrapBaseFile ( ) ) ) ) ) . collect ( Collectors . toList ( ) ) , cleanerParallelism ) . mapPartitionsToPair ( deleteFilesFunc ( table ) ) . reduceByKey ( PartitionCleanStat :: merge ) . collect ( ) ; Map < String , PartitionCleanStat > partitionCleanStatsMap = partitionCleanStats . stream ( ) . collect ( Collectors . toMap ( Tuple2 :: _1 , Tuple2 :: _2 ) ) ; return cleanerPlan . getFilePathsToBeDeletedPerPartition ( ) . keySet ( ) . stream ( ) . map ( partitionPath -> { PartitionCleanStat partitionCleanStat = partitionCleanStatsMap . containsKey ( partitionPath ) ? partitionCleanStatsMap . get ( partitionPath ) : new PartitionCleanStat ( partitionPath ) ; HoodieActionInstant actionInstant = cleanerPlan . getEarliestInstantToRetain ( ) ; return HoodieCleanStat . newBuilder ( ) . withPolicy ( config . getCleanerPolicy ( ) ) . withPartitionPath ( partitionPath ) . withEarliestCommitRetained ( Option . ofNullable ( actionInstant != null ? new HoodieInstant ( HoodieInstant . State . valueOf ( actionInstant . getState ( ) ) , actionInstant . getAction ( ) , actionInstant . getTimestamp ( ) ) : null ) ) . withDeletePathPattern ( partitionCleanStat . deletePathPatterns ( ) ) . withSuccessfulDeletes ( partitionCleanStat . successDeleteFiles ( ) ) . withFailedDeletes ( partitionCleanStat . failedDeleteFiles ( ) ) . withDeleteBootstrapBasePathPatterns ( partitionCleanStat . getDeleteBootstrapBasePathPatterns ( ) ) . withSuccessfulDeleteBootstrapBaseFiles ( partitionCleanStat . getSuccessfulDeleteBootstrapBaseFiles ( ) ) . withFailedDeleteBootstrapBaseFiles ( partitionCleanStat . getFailedDeleteBootstrapBaseFiles ( ) ) . build ( ) ; } ) . collect ( Collectors . toList ( ) ) ; }
Performs cleaning of partition paths according to cleaning policy and returns the number of files cleaned.
Performs cleaning of partition paths according to cleaning policy and returns the number of files cleaned .
// Return PartitionCleanStat for each partition passed.