什么是Clustering

开门见山，Clustering主要有两个作用：数据小文件合并和重排序。
当数据写入Hudi表时，为了提高写入效率和存储利用率，可能会产生大量小文件。Hudi的Clustering机制允许在后台周期性地将这些小文件合并成大文件，从而减少存储碎片和元数据管理开销，提高查询性能。
Clustering过程可以重新组织和排序数据，依据用户指定的列进行排序，这样能提升相关查询的性能，比如范围扫描或者JOIN操作，通过预排序的数据，查询引擎能够更高效地处理查询请求。
本篇分析clustering的源代码，包含clustering执行计划的创建和根据计划执行clustering过程两个部分。

创建clustering执行计划

创建执行计划位于ClusteringPlanActionExecutor类的execute方法，代码如下所示：

  @Override  
  public Option<HoodieClusteringPlan> execute() {  
    // 创建执行计划
    Option<HoodieClusteringPlan> planOption = createClusteringPlan();  
    // 如果计划创建成功（可能存在没有file slice需要cluster的情况）
    if (planOption.isPresent()) {  
      // 创建clustering instant
      // clustering instant的类型是replace commit，意味这clustering之后的数据文件替换掉先前的
      HoodieInstant clusteringInstant =  
          new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime);  
      try {  
        HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()  
            .setOperationType(WriteOperationType.CLUSTER.name())  
            .setExtraMetadata(extraMetadata.orElse(Collections.emptyMap()))  
            .setClusteringPlan(planOption.get())  
            .build();  
        // 添加到pending commit中
        table.getActiveTimeline().saveToPendingReplaceCommit(clusteringInstant,  
            TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));  
      } catch (IOException ioe) {  
        throw new HoodieIOException("Exception scheduling clustering", ioe);  
      }  
    }  
  
    return planOption;  
  }  
}

该方法创建clustering执行计划，然后再创建一个pending replace commit。因为clutering完成之后，新生成的数据文件会替换掉原有的数据文件，因此对应的commit类型为replace。
继续分析createClusteringPlan方法。其中首先判断是否满足可执行clustering的条件，然后获取配置的clustering策略类，创建clustering计划。
Clustering并不是说每次schedule都必须要执行。为了效率clustering要求至少要经过N次commit之后，才会schedule。此限制通过配置项hoodie.clustering.inline.max.commits或hoodie.clustering.async.max.commits(分别对应inline和异步)来控制。如果满足clustering条件，通过hoodie.clustering.plan.strategy.class配置的策略类生成执行计划。
代码如下所示：

protected Option<HoodieClusteringPlan> createClusteringPlan() {  
  LOG.info("Checking if clustering needs to be run on " + config.getBasePath()); 
  // 获取上一次clustering对应的instant 
  Option<HoodieInstant> lastClusteringInstant = table.getActiveTimeline().getLastClusterCommit();  
  // 获取上次clustering之后提交的次数
  int commitsSinceLastClustering = table.getActiveTimeline().getCommitsTimeline().filterCompletedInstants()  
      .findInstantsAfter(lastClusteringInstant.map(HoodieInstant::getTimestamp).orElse("0"), Integer.MAX_VALUE)  
      .countInstants();  
  // 读取hoodie.clustering.inline.max.commits配置，默认为4
  // 该配置项表示在上次clustering之后至少需要经历几次commit才能schedule下一次clustering
  // 这里处理inline clustering的配置
  if (config.inlineClusteringEnabled() && config.getInlineClusterMaxCommits() > commitsSinceLastClustering) {  
    LOG.warn("Not scheduling inline clustering as only " + commitsSinceLastClustering  
        + " commits was found since last clustering " + lastClusteringInstant + ". Waiting for "  
        + config.getInlineClusterMaxCommits());  
    return Option.empty();  
  }  
  // 同上，但这里处理异步clustering的配置
  // 配置项为hoodie.clustering.async.max.commits，默认值4
  if ((config.isAsyncClusteringEnabled() || config.scheduleInlineClustering()) && config.getAsyncClusterMaxCommits() > commitsSinceLastClustering) {  
    LOG.warn("Not scheduling async clustering as only " + commitsSinceLastClustering  
        + " commits was found since last clustering " + lastClusteringInstant + ". Waiting for "  
        + config.getAsyncClusterMaxCommits());  
    return Option.empty();  
  }  
  
  LOG.info("Generating clustering plan for table " + config.getBasePath());  
  // 加载clustering策略类，对应配置项hoodie.clustering.plan.strategy.class
  // 默认为SparkSizeBasedClusteringPlanStrategy
  ClusteringPlanStrategy strategy = (ClusteringPlanStrategy) ReflectionUtils.loadClass(  
      ClusteringPlanStrategy.checkAndGetClusteringPlanStrategy(config),  
          new Class<?>[] {HoodieTable.class, HoodieEngineContext.class, HoodieWriteConfig.class}, table, context, config);  
  // 生成clustering计划
  return strategy.generateClusteringPlan();  
}

接着我们聚焦默认的策略SparkSizeBasedClusteringPlanStrategy。该策略根据文件大小来决定文件数据是否参与clustering。分析clustering计划生成步骤。
generateClusteringPlan方法位于SparkSizeBasedClusteringPlanStrategy的父类PartitionAwareClusteringPlanStrategy中。该方法根据hoodie.clustering.plan.strategy.partition.selected，hoodie.clustering.plan.strategy.partition.regex.pattern和hoodie.clustering.plan.partition.filter.mode条件过滤出符合要求的partition path。获取它们包含的file slice。从这些file slice中筛选出小文件(小于hoodie.clustering.plan.strategy.small.file.limit的文件)。将这些按照clutering要求的group大小(hoodie.clustering.plan.strategy.max.bytes.per.group)，分成若干个group。Group数量上限为hoodie.clustering.plan.strategy.max.num.groups。此步骤对应小文件合并功能。
代码如下所示：

@Override  
public Option<HoodieClusteringPlan> generateClusteringPlan() {  
  if (!checkPrecondition()) {  
    return Option.empty();  
  }  
  // 获取metaclient，用来操作metadata
  HoodieTableMetaClient metaClient = getHoodieTable().getMetaClient();  
  LOG.info("Scheduling clustering for " + metaClient.getBasePath());  
  // 获取写配置
  HoodieWriteConfig config = getWriteConfig();  
  // 读取配置项hoodie.clustering.plan.strategy.partition.selected
  // 确定在哪些分区运行clustering
  String partitionSelected = config.getClusteringPartitionSelected();  
  LOG.info("Scheduling clustering partitionSelected: " + partitionSelected);  
  List<String> partitionPaths;  

  // 如果没有配置
  if (StringUtils.isNullOrEmpty(partitionSelected)) {  
    // get matched partitions if set  
    // 读取hoodie.clustering.plan.strategy.partition.regex.pattern配置
    // 获取正则表达式匹配的partition path
    partitionPaths = getRegexPatternMatchedPartitions(config, FSUtils.getAllPartitionPaths(getEngineContext(), config.getMetadataConfig(), metaClient.getBasePath()));  
    // filter the partition paths if needed to reduce list status  
  } else {  
    // 如果配置了partitionSelected，优先这个配置
    partitionPaths = Arrays.asList(partitionSelected.split(","));  
  }  
  // 过滤需要clustering的分区
  // 过滤策略对应配置项hoodie.clustering.plan.partition.filter.mode
  // 可用策略为NONE，RECENT_DAYS，SELECTED_PARTITIONS和DAY_ROLLING
  partitionPaths = filterPartitionPaths(partitionPaths);  
  LOG.info("Scheduling clustering partitionPaths: " + partitionPaths);  

  // 如果所有的分区都被排除了，返回空
  if (partitionPaths.isEmpty()) {  
    // In case no partitions could be picked, return no clustering plan  
    return Option.empty();  
  }  

  // 排除掉分区中已经要做clustering的file group（pending状态）
  // 筛选出小文件
  // 决定小文件判断阈值的配置项为hoodie.clustering.plan.strategy.small.file.limit
  // 将其映射为HoodieClusteringGroup
  // 映射逻辑后面分析
  List<HoodieClusteringGroup> clusteringGroups = getEngineContext()  
      .flatMap(  
          partitionPaths,  
          partitionPath -> {  
            List<FileSlice> fileSlicesEligible = getFileSlicesEligibleForClustering(partitionPath).collect(Collectors.toList());  
            return buildClusteringGroupsForPartition(partitionPath, fileSlicesEligible).limit(getWriteConfig().getClusteringMaxNumGroups());  
          },  
          partitionPaths.size())  
      .stream()  
      .limit(getWriteConfig().getClusteringMaxNumGroups())  
      .collect(Collectors.toList());  
  
  if (clusteringGroups.isEmpty()) {  
    LOG.warn("No data available to cluster");  
    return Option.empty();  
  }  
  // 构造cluster策略
  HoodieClusteringStrategy strategy = HoodieClusteringStrategy.newBuilder()  
      .setStrategyClassName(getWriteConfig().getClusteringExecutionStrategyClass())  
      .setStrategyParams(getStrategyParams())  
      .build();  

  // 构造clustering计划
  return Option.of(HoodieClusteringPlan.newBuilder()  
      .setStrategy(strategy)  
      .setInputGroups(clusteringGroups)  
      .setExtraMetadata(getExtraMetadata())  
      .setVersion(getPlanVersion())  
      .setPreserveHoodieMetadata(true)  
      .build());  
}

上面的filterPartitionPaths通过配置的hoodie.clustering.plan.partition.filter.mode过滤出所需的partition。具有有如下选项：

• NONE: 不过滤，返回所有partition path。
• RECENT_DAYS: 按照partition path倒序排序。跳过hoodie.clustering.plan.strategy.daybased.skipfromlatest.partitions个partition，返回hoodie.clustering.plan.strategy.daybased.lookback.partitions个partition。如果partition path是日期，可以实现过滤出最近N天的数据。
• SELECTED_PARTITIONS: 获取hoodie.clustering.plan.strategy.cluster.begin.partition和hoodie.clustering.plan.strategy.cluster.end.partition之间的分区。
• DAY_ROLLING: 每次clustering一部分分区。如果分区的index对24取余等于排期时候当前时间的小时数，则该分区需要clustering。

buildClusteringGroupsForPartition方法将筛选出的file slice按照从小到大排序。然后按照clustering配置的group size和group数量条件，合并为clustering group。

protected Stream<HoodieClusteringGroup> buildClusteringGroupsForPartition(String partitionPath, List<FileSlice> fileSlices) {  
  // 获取写入配置
  HoodieWriteConfig writeConfig = getWriteConfig();  
  
  List<Pair<List<FileSlice>, Integer>> fileSliceGroups = new ArrayList<>();  
  List<FileSlice> currentGroup = new ArrayList<>();  
  
  // Sort fileSlices before dividing, which makes dividing more compact  
  // file slice按照base file大小排序，如果文件不存在，按照最大大小排序
  List<FileSlice> sortedFileSlices = new ArrayList<>(fileSlices);  
  sortedFileSlices.sort((o1, o2) -> (int)  
      ((o2.getBaseFile().isPresent() ? o2.getBaseFile().get().getFileSize() : writeConfig.getParquetMaxFileSize())  
          - (o1.getBaseFile().isPresent() ? o1.getBaseFile().get().getFileSize() : writeConfig.getParquetMaxFileSize())));  
  
  long totalSizeSoFar = 0;  
  
  for (FileSlice currentSlice : sortedFileSlices) {  
    // 遍历所有file slice
    // 获取当前file slice的大小，如果文件不存在，获取大小上限
    long currentSize = currentSlice.getBaseFile().isPresent() ? currentSlice.getBaseFile().get().getFileSize() : writeConfig.getParquetMaxFileSize();  
    // check if max size is reached and create new group, if needed.  
    // 如果本次累积的文件大小大于hoodie.clustering.plan.strategy.max.bytes.per.group
    // 并且当前group不为空
    if (totalSizeSoFar + currentSize > writeConfig.getClusteringMaxBytesInGroup() && !currentGroup.isEmpty()) {  
      // totalSizeSoFar除以hoodie.clustering.plan.strategy.target.file.max.bytes向上取整
      // 计算出输出组编号
      int numOutputGroups = getNumberOfOutputFileGroups(totalSizeSoFar, writeConfig.getClusteringTargetFileMaxBytes());  
      LOG.info("Adding one clustering group " + totalSizeSoFar + " max bytes: "  
          + writeConfig.getClusteringMaxBytesInGroup() + " num input slices: " + currentGroup.size() + " output groups: " + numOutputGroups);  
      // 加入到fileSliceGroups集合中，保存结果
      // 保存了输出组组和输出组编号
      fileSliceGroups.add(Pair.of(currentGroup, numOutputGroups));  
      // 结果保存之后，清零currentGroup和totalSizeSoFar
      currentGroup = new ArrayList<>();  
      totalSizeSoFar = 0;  
  
      // if fileSliceGroups's size reach the max group, stop loop  
      // 检查file group个数是否超过了hoodie.clustering.plan.strategy.max.num.groups
      // 超过的话退出循环，本次不再处理后面的file slice
      if (fileSliceGroups.size() >= writeConfig.getClusteringMaxNumGroups()) {  
        LOG.info("Having generated the maximum number of groups : " + writeConfig.getClusteringMaxNumGroups());  
        break;  
      }  
    }  
  
    // Add to the current file-group  
    // 加入到当前文件组
    currentGroup.add(currentSlice);  
    // assume each file group size is ~= parquet.max.file.size  
    // 累积大小到totalSizeSoFar变量
    totalSizeSoFar += currentSize;  
  }  
  
  if (!currentGroup.isEmpty()) {  
    // 处理最后一个output group
    // shouldClusteringSingleGroup在下面两个配置项任意一个启用的时候为true
    // 表示只有一个输出文件组的话，也clustering
    // hoodie.clustering.plan.strategy.sort.columns
    // hoodie.clustering.plan.strategy.single.group.clustering.enabled
    if (currentGroup.size() > 1 || writeConfig.shouldClusteringSingleGroup()) {  
      int numOutputGroups = getNumberOfOutputFileGroups(totalSizeSoFar, writeConfig.getClusteringTargetFileMaxBytes());  
      LOG.info("Adding final clustering group " + totalSizeSoFar + " max bytes: "  
          + writeConfig.getClusteringMaxBytesInGroup() + " num input slices: " + currentGroup.size() + " output groups: " + numOutputGroups);  
      fileSliceGroups.add(Pair.of(currentGroup, numOutputGroups));  
    }  
  }
  // 构造并返回fileSliceGroups
  return fileSliceGroups.stream().map(fileSliceGroup ->  
    HoodieClusteringGroup.newBuilder()  
        .setSlices(getFileSliceInfo(fileSliceGroup.getLeft()))  
        .setNumOutputFileGroups(fileSliceGroup.getRight())  
        .setMetrics(buildMetrics(fileSliceGroup.getLeft()))  
        .build());
}

到此为止clustering计划生成部分分析完毕。

根据执行计划执行clustering

Clustering的执行开始于BaseHoodieWriteClient::cluster。
在clustering之前，首先执行preWrite操作。

public HoodieWriteMetadata<O> cluster(String clusteringInstant, boolean shouldComplete) {  
  // 创建hudi table，根据引擎(Spark/Flink)和表类型(MOR/COW)的不同，有多种实现类
  HoodieTable table = createTable(config, context.getHadoopConf().get());
  // 执行写入前操作，包含：
  // inflight和requested instant去掉本次instant
  // 启动clean和archive服务（如果开启的话）
  preWrite(clusteringInstant, WriteOperationType.CLUSTER, table.getMetaClient());
  // 执行clutering
  return tableServiceClient.cluster(clusteringInstant, shouldComplete);
}

接着是BaseHoodieTableServiceClient::cluster方法。该方法检测当前clustering是否已经pending，配置监控，执行clustering并返回clustering执行结果元数据。

public HoodieWriteMetadata<O> cluster(String clusteringInstant, boolean shouldComplete) {  
  // 同上个方法，获取table
  HoodieTable<?, I, ?, T> table = createTable(config, context.getHadoopConf().get());  
  HoodieTimeline pendingClusteringTimeline = table.getActiveTimeline().filterPendingReplaceTimeline();  
  HoodieInstant inflightInstant = HoodieTimeline.getReplaceCommitInflightInstant(clusteringInstant);  
  // 检查本次cluster是否已经pending状态。如果是，需要回滚
  if (pendingClusteringTimeline.containsInstant(inflightInstant)) {  
    table.rollbackInflightClustering(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false));  
    table.getMetaClient().reloadActiveTimeline();  
  }  
  // cluster时长计时器监控
  clusteringTimer = metrics.getClusteringCtx();  
  LOG.info("Starting clustering at " + clusteringInstant);  
  // 调用table的cluster服务
  HoodieWriteMetadata<T> writeMetadata = table.cluster(context, clusteringInstant);
  // 转换metadata到对应计算引擎格式
  HoodieWriteMetadata<O> clusteringMetadata = convertToOutputMetadata(writeMetadata);  
  // Validation has to be done after cloning. if not, it could result in referencing the write status twice which means clustering could get executed twice.  
  // 检查cluster写入状态不能为空
  validateClusteringCommit(clusteringMetadata, clusteringInstant, table);  
  
  // Publish file creation metrics for clustering.  
  // 读取并返回监控信息
  if (config.isMetricsOn()) {  
    clusteringMetadata.getWriteStats()  
        .ifPresent(hoodieWriteStats -> hoodieWriteStats.stream()  
            .filter(hoodieWriteStat -> hoodieWriteStat.getRuntimeStats() != null)  
            .map(hoodieWriteStat -> hoodieWriteStat.getRuntimeStats().getTotalCreateTime())  
            .forEach(metrics::updateClusteringFileCreationMetrics));  
  }  
  
  // TODO : Where is shouldComplete used ?  
  if (shouldComplete && clusteringMetadata.getCommitMetadata().isPresent()) {  
    completeClustering((HoodieReplaceCommitMetadata) clusteringMetadata.getCommitMetadata().get(), table, clusteringInstant, Option.ofNullable(convertToWriteStatus(writeMetadata)));  
  }  
  return clusteringMetadata;  
}

以Spark为例，我们查看COW表的HoodieSparkCopyOnWriteTable::cluster逻辑。

public HoodieWriteMetadata<HoodieData<WriteStatus>> cluster(HoodieEngineContext context,  
                                                         String clusteringInstantTime) {  
  return new SparkExecuteClusteringCommitActionExecutor<>(context, config, this, clusteringInstantTime).execute();  
}

此逻辑交由SparkExecuteClusteringCommitActionExecutor执行。继续分析SparkExecuteClusteringCommitActionExecutor::execute方法，它调用了BaseCommitActionExecutor::executeClustering方法。

@Override  
public HoodieWriteMetadata<HoodieData<WriteStatus>> execute() {  
  return executeClustering(clusteringPlan);  
}

BaseCommitActionExecutor::executeClustering该方法反射加载hoodie.clustering.execution.strategy.class配置项对应的clustering策略（默认为SparkSortAndSizeExecutionStrategy），然后执行clustering。

protected HoodieWriteMetadata<HoodieData<WriteStatus>> executeClustering(HoodieClusteringPlan clusteringPlan) {  
  // 创建instant
  HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(instantTime);  
  // Mark instant as clustering inflight  
  // 标记instant为inflight状态
  table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());  
  table.getMetaClient().reloadActiveTimeline();  
  
  // Disable auto commit. Strategy is only expected to write data in new files.  
  // 禁用自动commit
  config.setValue(HoodieWriteConfig.AUTO_COMMIT_ENABLE, Boolean.FALSE.toString());  

  // 添加_hoodie_commit_time等5个元数据字段到schema中
  final Schema schema = HoodieAvroUtils.addMetadataFields(new Schema.Parser().parse(config.getSchema()));  
  // 加载hoodie.clustering.execution.strategy.class配置项对应的clustering策略类
  // 执行它的performClustering方法
  // 对于默认的配置，clustering策略类为SparkSortAndSizeExecutionStrategy
  HoodieWriteMetadata<HoodieData<WriteStatus>> writeMetadata = (  
      (ClusteringExecutionStrategy<T, HoodieData<HoodieRecord<T>>, HoodieData<HoodieKey>, HoodieData<WriteStatus>>)  
          ReflectionUtils.loadClass(config.getClusteringExecutionStrategyClass(),  
              new Class<?>[] {HoodieTable.class, HoodieEngineContext.class, HoodieWriteConfig.class}, table, context, config))  
      .performClustering(clusteringPlan, schema, instantTime);  
  // 获取写入状态
  HoodieData<WriteStatus> writeStatusList = writeMetadata.getWriteStatuses();  
  // 更新表索引，更新数据所在位置
  HoodieData<WriteStatus> statuses = updateIndex(writeStatusList, writeMetadata);  
  // 持久化保存
  statuses.persist(config.getString(WRITE_STATUS_STORAGE_LEVEL_VALUE), context, HoodieData.HoodieDataCacheKey.of(config.getBasePath(), instantTime));  
  // triggers clustering.  
  // 更新writeMetadata中的writestats
  writeMetadata.setWriteStats(statuses.map(WriteStatus::getStat).collectAsList());  
// 获取clustering操作的数据文件file id和partition path
writeMetadata.setPartitionToReplaceFileIds(getPartitionToReplacedFileIds(clusteringPlan, writeMetadata)); 
// 提交修改的writeMetadata，clustering对后续操作生效
  commitOnAutoCommit(writeMetadata);  
  if (!writeMetadata.getCommitMetadata().isPresent()) {  
    HoodieCommitMetadata commitMetadata = CommitUtils.buildMetadata(writeMetadata.getWriteStats().get(), writeMetadata.getPartitionToReplaceFileIds(),  
        extraMetadata, operationType, getSchemaToStoreInCommit(), getCommitActionType());  
    writeMetadata.setCommitMetadata(Option.of(commitMetadata));  
  }  
  return writeMetadata;  
}

Clustering的执行细节位于策略类中。我们这里分析默认的策略类SparkSortAndSizeExecutionStrategy::performClustering方法。该方法位于父类MultipleSparkJobExecutionStrategy::performClustering中。该方法使用线程池，一个线程处理一个input group（对应执行计划中提到的clustering group），但线程数不能超过配置的最大值。

@Override  
public HoodieWriteMetadata<HoodieData<WriteStatus>> performClustering(final HoodieClusteringPlan clusteringPlan, final Schema schema, final String instantTime) {  
  JavaSparkContext engineContext = HoodieSparkEngineContext.getSparkContext(getEngineContext());  
  // 是否保留元数据，默认为true
  boolean shouldPreserveMetadata = Option.ofNullable(clusteringPlan.getPreserveHoodieMetadata()).orElse(true);  
  // 使用专门的线程执行clustering。创建clustering线程池
  // 取InputGroups数量（plan中clustering生成file group的数量）
  // 最大值为hoodie.clustering.max.parallelism，最大值默认15
  ExecutorService clusteringExecutorService = Executors.newFixedThreadPool(  
      Math.min(clusteringPlan.getInputGroups().size(), writeConfig.getClusteringMaxParallelism()),  
      new CustomizedThreadFactory("clustering-job-group", true));  
  try {  
    // execute clustering for each group async and collect WriteStatus  
    // 在线程池中执行clustering，获取执行结果
    Stream<HoodieData<WriteStatus>> writeStatusesStream = FutureUtils.allOf(  
            clusteringPlan.getInputGroups().stream()  
                .map(inputGroup -> {  
 // hoodie.datasource.write.row.writer.enable如果为true，使用Spark原生的Row类型，避免类型转换引发的额外代价
                  if (getWriteConfig().getBooleanOrDefault("hoodie.datasource.write.row.writer.enable", true)) {  
                    return runClusteringForGroupAsyncAsRow(inputGroup,  
                        clusteringPlan.getStrategy().getStrategyParams(),  
                        shouldPreserveMetadata,  
                        instantTime,  
                        clusteringExecutorService);  
                  }  
                  return runClusteringForGroupAsync(inputGroup,  
                      clusteringPlan.getStrategy().getStrategyParams(),  
                      shouldPreserveMetadata,  
                      instantTime,  
                      clusteringExecutorService);  
                })  
                .collect(Collectors.toList()))  
        .join()  
        .stream();  
    JavaRDD<WriteStatus>[] writeStatuses = convertStreamToArray(writeStatusesStream.map(HoodieJavaRDD::getJavaRDD));  
    JavaRDD<WriteStatus> writeStatusRDD = engineContext.union(writeStatuses);  
  
    HoodieWriteMetadata<HoodieData<WriteStatus>> writeMetadata = new HoodieWriteMetadata<>();  
    writeMetadata.setWriteStatuses(HoodieJavaRDD.of(writeStatusRDD));  
    return writeMetadata;  
  } finally {  
    clusteringExecutorService.shutdown();  
  }  
}

我们继续分析默认配置的执行路线ClusteringPlanActionExecutor::runClusteringForGroupAsyncAsRow。该方法获取到所有需要clustering的数据到Spark的dataset，读取表schema和各个file id从属的partition path的对应关系。然后执行clustering。

private CompletableFuture<HoodieData<WriteStatus>> runClusteringForGroupAsyncAsRow(HoodieClusteringGroup clusteringGroup,  
                                                                                   Map<String, String> strategyParams,  
                                                                                   boolean shouldPreserveHoodieMetadata,  
                                                                                   String instantTime,  
                                                                                   ExecutorService clusteringExecutorService) {  
  return CompletableFuture.supplyAsync(() -> {  
    // 获取spark context
    JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(getEngineContext());  
    // 转换所有clustering涉及到的数据为Spark DataSet
    Dataset<Row> inputRecords = readRecordsForGroupAsRow(jsc, clusteringGroup, instantTime);  
    // 获取带有元数据字段的schema
    Schema readerSchema = HoodieAvroUtils.addMetadataFields(new Schema.Parser().parse(getWriteConfig().getSchema()));  
    // 转换clustering的file slice为HoodieFileGroupId
    // 保存的是partition path和file id的对应关系
    List<HoodieFileGroupId> inputFileIds = clusteringGroup.getSlices().stream()  
        .map(info -> new HoodieFileGroupId(info.getPartitionPath(), info.getFileId()))  
        .collect(Collectors.toList());  
    // 执行clustering
    return performClusteringWithRecordsAsRow(inputRecords, clusteringGroup.getNumOutputFileGroups(), instantTime, strategyParams, readerSchema, inputFileIds, shouldPreserveHoodieMetadata,  
        clusteringGroup.getExtraMetadata());  
  }, clusteringExecutorService);  
}

SparkSortAndSizeExecutionStrategy::performClusteringWithRecordsAsRow方法获取分区器，将数据重新排序，最后使用批量插入的方式，写回parquet文件。

@Override  
public HoodieData<WriteStatus> performClusteringWithRecordsAsRow(Dataset<Row> inputRecords,  
                                                                 int numOutputGroups,  
                                                                 String instantTime, Map<String, String> strategyParams,  
                                                                 Schema schema,  
                                                                 List<HoodieFileGroupId> fileGroupIdList,  
                                                                 boolean shouldPreserveHoodieMetadata,  
                                                                 Map<String, String> extraMetadata) {  
  LOG.info("Starting clustering for a group, parallelism:" + numOutputGroups + " commit:" + instantTime);  
  // 生成写入配置，clustering输出多少个file group就配置多少个bulk insert并行度
  HoodieWriteConfig newConfig = HoodieWriteConfig.newBuilder()  
      .withBulkInsertParallelism(numOutputGroups)  
      .withProps(getWriteConfig().getProps()).build();  
  // 配置最大parquet文件大小为clustering目标文件最大上限
  // 对应配置项为hoodie.clustering.plan.strategy.target.file.max.bytes
  newConfig.setValue(HoodieStorageConfig.PARQUET_MAX_FILE_SIZE, String.valueOf(getWriteConfig().getClusteringTargetFileMaxBytes()));  

  // 获取分区器
  BulkInsertPartitioner<Dataset<Row>> partitioner = getRowPartitioner(strategyParams, schema);  
  // 使用分区器分区数据（数据重新排序）
  Dataset<Row> repartitionedRecords = partitioner.repartitionRecords(inputRecords, numOutputGroups);  
  // 将重排序之后的数据批量插入
  return HoodieDatasetBulkInsertHelper.bulkInsert(repartitionedRecords, instantTime, getHoodieTable(), newConfig,  
      partitioner.arePartitionRecordsSorted(), shouldPreserveHoodieMetadata);  
}

接下来分析的重点是clustering的另一个功能：将数据重排序。因此重点是分区器和分区器重排序的逻辑。获取分区器的逻辑位于它的父类MultipleSparkJobExecutionStrategy::getRowPartitioner中。代码如下：

private <I> BulkInsertPartitioner<I> getPartitioner(Map<String, String> strategyParams,  
                                                    Schema schema,  
                                                    boolean isRowPartitioner) { 
  // 获取排序字段配置项
  // 对应的配置项为hoodie.clustering.plan.strategy.sort.columns 
  // 使用逗号分隔
  Option<String[]> orderByColumnsOpt =  
      Option.ofNullable(strategyParams.get(PLAN_STRATEGY_SORT_COLUMNS.key()))  
          .map(listStr -> listStr.split(","));  
  
  return orderByColumnsOpt.map(orderByColumns -> {  
    // 获取hoodie.layout.optimize.strategy配置，字段可使用zorder或者hilbert曲线排序或者linear线性排序
    HoodieClusteringConfig.LayoutOptimizationStrategy layoutOptStrategy = getWriteConfig().getLayoutOptimizationStrategy();  
    switch (layoutOptStrategy) {  
      case ZORDER:  
      case HILBERT:  
        return isRowPartitioner  
            ? new RowSpatialCurveSortPartitioner(getWriteConfig())  
            : new RDDSpatialCurveSortPartitioner((HoodieSparkEngineContext) getEngineContext(), orderByColumns, layoutOptStrategy,  
            getWriteConfig().getLayoutOptimizationCurveBuildMethod(), HoodieAvroUtils.addMetadataFields(schema), recordType);  
      case LINEAR:  
        return isRowPartitioner  
            ? new RowCustomColumnsSortPartitioner(orderByColumns, getWriteConfig())  
            : new RDDCustomColumnsSortPartitioner(orderByColumns, HoodieAvroUtils.addMetadataFields(schema), getWriteConfig());  
      default:  
        throw new UnsupportedOperationException(String.format("Layout optimization strategy '%s' is not supported", layoutOptStrategy));  
    }  
  }).orElseGet(() -> isRowPartitioner  
      ? BulkInsertInternalPartitionerWithRowsFactory.get(getWriteConfig(), getHoodieTable().isPartitioned(), true)  
      : BulkInsertInternalPartitionerFactory.get(getHoodieTable(), getWriteConfig(), true));  
}

对于使用Spark原生Row类型的情况，isRowPartitioner为true。如果使用ZORDER或者HILBERT排序策略，使用RowSpatialCurveSortPartitioner，LINEAR排序策略对应的是RowCustomColumnsSortPartitioner。
接下来我们分别分析这两个partitioner是如何对数据重排序的。
首先是RowSpatialCurveSortPartitioner::repartitionRecords，代码如下：

@Override  
public Dataset<Row> repartitionRecords(Dataset<Row> records, int outputPartitions) {  
  return reorder(records, outputPartitions);  
}

repartitionRecords调用了reorder方法。

protected Dataset<Row> reorder(Dataset<Row> dataset, int numOutputGroups) { 
  // 检查排序字段配置
  if (orderByColumns.length == 0) {  
    // No-op  
    return dataset;  
  }  
  
  List<String> orderedCols = Arrays.asList(orderByColumns);  
  // curveCompositionStrategyType默认为DIRECT
  switch (curveCompositionStrategyType) {  
    case DIRECT:  
      return SpaceCurveSortingHelper.orderDataFrameByMappingValues(dataset, layoutOptStrategy, orderedCols, numOutputGroups);  
    case SAMPLE:  
      return SpaceCurveSortingHelper.orderDataFrameBySamplingValues(dataset, layoutOptStrategy, orderedCols, numOutputGroups);  
    default:  
      throw new UnsupportedOperationException(String.format("Unsupported space-curve curve building strategy (%s)", curveCompositionStrategyType));  
  }  
}

SpatialCurveCompositionStrategyType中SAMPLE的数据排序分布效果较DIRECT更好，但是执行速度更慢。默认配置的是DIRECT类型。
接下来分析DIRECT类型处理方式，对应的是SpaceCurveSortingHelper::orderDataFrameByMappingValues。该方法首先判断排序字段配置的合法性。然后将数据按照排序字段，使用Z曲线或者是Hilbert曲线重排序。

public static Dataset<Row> orderDataFrameByMappingValues(  
    Dataset<Row> df,  
    HoodieClusteringConfig.LayoutOptimizationStrategy layoutOptStrategy,  
    List<String> orderByCols,  
    int targetPartitionCount  
) {  
  // 获取字段名称和StructField的对应关系
  Map<String, StructField> columnsMap =  
      Arrays.stream(df.schema().fields())  
          .collect(Collectors.toMap(StructField::name, Function.identity()));  
  // 检查排序字段是否出现在schema中
  List<String> checkCols =  
      orderByCols.stream()  
          .filter(columnsMap::containsKey)  
          .collect(Collectors.toList());  
  // 如果没有，说明排序字段配置有误，跳过不再继续执行
  if (orderByCols.size() != checkCols.size()) {  
    LOG.error(String.format("Trying to ordering over a column(s) not present in the schema (%s); skipping", CollectionUtils.diff(orderByCols, checkCols)));  
    return df;  
  }  
  
  // In case when there's just one column to be ordered by, we can skip space-curve  
  // ordering altogether (since it will match linear ordering anyway) 
  // 如果排序字段只有一个，没必要使用空间曲线方式排序，直接使用Spark排序
  if (orderByCols.size() == 1) {  
    String orderByColName = orderByCols.get(0);  
    LOG.debug(String.format("Single column to order by (%s), skipping space-curve ordering", orderByColName));  
  
    // TODO validate if we need Spark to re-partition  
    return df.repartitionByRange(targetPartitionCount, new Column(orderByColName));  
  }  
  // 字段个数
  int fieldNum = df.schema().fields().length;  

  // 返回排序字段对应的index和字段信息对应关系
  Map<Integer, StructField> fieldMap =  
      orderByCols.stream()  
          .collect(  
              Collectors.toMap(e -> Arrays.asList(df.schema().fields()).indexOf(columnsMap.get(e)), columnsMap::get));  
  
  JavaRDD<Row> sortedRDD;  
  // 根据布局优化策略，排序RDD
  switch (layoutOptStrategy) {  
    case ZORDER:  
      sortedRDD = createZCurveSortedRDD(df.toJavaRDD(), fieldMap, fieldNum, targetPartitionCount);  
      break;  
    case HILBERT:  
      sortedRDD = createHilbertSortedRDD(df.toJavaRDD(), fieldMap, fieldNum, targetPartitionCount);  
      break;  
    default:  
      throw new UnsupportedOperationException(String.format("Not supported layout-optimization strategy (%s)", layoutOptStrategy));  
  }  
  
  // Compose new {@code StructType} for ordered RDDs  
  // 为排序后的RDD创建StructType（schema）
  StructType newStructType = composeOrderedRDDStructType(df.schema());  

  // 返回dataset
  return df.sparkSession()  
      .createDataFrame(sortedRDD, newStructType)  
      .drop("Index");  
}

我们先看第一种情况，Z曲线排序。方法位于SpaceCurveSortingHelper::createZCurveSortedRDD。
该方法将多个排序字段的值映射为8字节内容（多的截取少的补充），然后每个字段的字节内容各取一位拼接到一起，然后再各取第二位拼接……一直循环，这个步骤称之为二进制数据交织(interleaving)。将交织之后的值作为一个字段，拼接在数据中。然后按照该字段的内容排序。

private static JavaRDD<Row> createZCurveSortedRDD(JavaRDD<Row> originRDD, Map<Integer, StructField> fieldMap, int fieldNum, int fileNum) {  
  return originRDD.map(row -> {  
    // 将数据中每个排序字段的值填充为8字节内容
    // 多的截取少的补充
    byte[][] zBytes = fieldMap.entrySet().stream()  
      .map(entry -> {  
        int index = entry.getKey();  
        StructField field = entry.getValue();  
        return mapColumnValueTo8Bytes(row, index, field.dataType());  
      })  
      .toArray(byte[][]::new);  

    // Interleave received bytes to produce Z-curve ordinal  
    // 将这些排序字段的值交织起来
    // 比如有A，B两个排序字段。A字段值取1位，然后取B字段值1位，然后A再取下一位，B取下一位，以此类推
    byte[] zOrdinalBytes = BinaryUtil.interleaving(zBytes, 8);  
    // 追加zOrdinalBytes到Row
    return appendToRow(row, zOrdinalBytes);  
  })  
    // 按照该字段的值(zOrdinalBytes，位于row的末尾，index正好是fieldNum)排序
    .sortBy(f -> new ByteArraySorting((byte[]) f.get(fieldNum)), true, fileNum);  
}

第二种情况为Hilbert曲线，对应方法为SpaceCurveSortingHelper::createHilbertSortedRDD。和ZOrder曲线排序处理逻辑基本相同，只是将Z曲线替换成了Hilbert曲线。

private static JavaRDD<Row> createHilbertSortedRDD(JavaRDD<Row> originRDD, Map<Integer, StructField> fieldMap, int fieldNum, int fileNum) {  
  // NOTE: Here {@code mapPartitions} is used to make sure Hilbert curve instance is initialized  
  //       only once per partition  
  return originRDD.mapPartitions(rows -> {  
    // 创建hilbert fieldMap个数维度曲线
    HilbertCurve hilbertCurve = HilbertCurve.bits(63).dimensions(fieldMap.size());  
    return new Iterator<Row>() {  
  
      @Override  
      public boolean hasNext() {  
        return rows.hasNext();  
      }  
  
      @Override  
      public Row next() {  
        Row row = rows.next();  
        // 将row中的排序字段值映射为long类型
        long[] longs = fieldMap.entrySet().stream()  
            .mapToLong(entry -> {  
              int index = entry.getKey();  
              StructField field = entry.getValue();  
              return mapColumnValueToLong(row, index, field.dataType());  
            })  
            .toArray();  
  
        // Map N-dimensional coordinates into position on the Hilbert curve  
        // 使用hilbert曲线索引上面的long值，结果作为后面的排序依据
        byte[] hilbertCurvePosBytes = HilbertCurveUtils.indexBytes(hilbertCurve, longs, 63);  
        return appendToRow(row, hilbertCurvePosBytes);  
      }  
    };  
  })  
      .sortBy(f -> new ByteArraySorting((byte[]) f.get(fieldNum)), true, fileNum);  
}

和空间曲线的方式相比，LINEAR线性排序显得较为简单。代码位于RowCustomColumnsSortPartitioner::repartitionRecords
。通过spark的sort算子按照配置的column排序。

@Override  
public Dataset<Row> repartitionRecords(Dataset<Row> records, int outputSparkPartitions) {  
  return records  
      .sort(Arrays.stream(sortColumnNames).map(Column::new).toArray(Column[]::new))  
      .coalesce(outputSparkPartitions);  
}

到这里位置分区器的逻辑分析完毕。