Apache Hudi索引实现分析（二）之HoodieGlobalBloomIndex

ApacheHudi

发布于 2021-04-13 11:06:52

6930

发布于 2021-04-13 11:06:52

文章被收录于专栏：ApacheHudi

1. 介绍

前面分析了Hudi默认的索引实现HoodieBloomIndex，其是基于分区记录所在文件，即分区路径+recordKey唯一即可，Hudi还提供了HoodieGlobalBloomIndex的实现，即全局索引实现，只需要recordKey唯一即可，下面分析其实现。

2. 分析

HoodieGlobalBloomIndex是HoodieBloomIndex的子类，其主要重写了父类的如下几个方法

// 加载分区下所有最新的文件
List<Tuple2<String, BloomIndexFileInfo>> loadInvolvedFiles(List<String> partitions, final JavaSparkContext jsc, final HoodieTable hoodieTable)

// 查找记录对应的文件
JavaRDD<Tuple2<String, HoodieKey>> explodeRecordRDDWithFileComparisons(final Map<String, List<BloomIndexFileInfo>> partitionToFileIndexInfo, JavaPairRDD<String, String> partitionRecordKeyPairRDD)

// 将位置信息推回至原始记录
protected JavaRDD<HoodieRecord<T>> tagLocationBacktoRecords(JavaPairRDD<HoodieKey, HoodieRecordLocation> keyLocationPairRDD, JavaRDD<HoodieRecord<T>> recordRDD)

对于加载分区下所有最新文件而言， loadInvolvedFiles核心代码如下

List<Tuple2<String, BloomIndexFileInfo>> loadInvolvedFiles(List<String> partitions, final JavaSparkContext jsc,
                                                             final HoodieTable hoodieTable) {
    HoodieTableMetaClient metaClient = hoodieTable.getMetaClient();
    try {
      // 获取所有分区
      List<String> allPartitionPaths = FSUtils.getAllPartitionPaths(metaClient.getFs(), metaClient.getBasePath(),
          config.shouldAssumeDatePartitioning());
      // 调用父类方法加载所有分区下最新数据文件
      return super.loadInvolvedFiles(allPartitionPaths, jsc, hoodieTable);
    } catch (IOException e) {
      throw new HoodieIOException("Failed to load all partitions", e);
    }
  }

首先会获取所有的分区路径，然后调用父类方法获取分区下最新数据文件。

对于查找记录对应的文件而言， explodeRecordRDDWithFileComparisons核心代码如下

JavaRDD<Tuple2<String, HoodieKey>> explodeRecordRDDWithFileComparisons(
      final Map<String, List<BloomIndexFileInfo>> partitionToFileIndexInfo,
      JavaPairRDD<String, String> partitionRecordKeyPairRDD) {
    // 使用索引过滤器，根据之前读取的最大和最小recordKey进行初始化
    IndexFileFilter indexFileFilter =
        config.getBloomIndexPruneByRanges() ? new IntervalTreeBasedGlobalIndexFileFilter(partitionToFileIndexInfo)
            : new ListBasedGlobalIndexFileFilter(partitionToFileIndexInfo);

    return partitionRecordKeyPairRDD.map(partitionRecordKeyPair -> {
      String recordKey = partitionRecordKeyPair._2();
      String partitionPath = partitionRecordKeyPair._1();
      // 获取匹配的文件和partition
      return indexFileFilter.getMatchingFilesAndPartition(partitionPath, recordKey).stream()
          .map(partitionFileIdPair -> new Tuple2<>(partitionFileIdPair.getRight(),
              new HoodieKey(recordKey, partitionFileIdPair.getLeft())))
          .collect(Collectors.toList());
    }).flatMap(List::iterator);
  }

可以看到和 HoodieBloomIndex#explodeRecordRDDWithFileComparisons处理逻辑类似，在使用索引过滤器获取所有匹配的文件和分区路径时，此时比较的是所有分区下的文件，不再是指定的分区路径。

对于将位置信息推回至原始记录而言， tagLocationBacktoRecords核心代码如下

protected JavaRDD<HoodieRecord<T>> tagLocationBacktoRecords(
      JavaPairRDD<HoodieKey, HoodieRecordLocation> keyLocationPairRDD, JavaRDD<HoodieRecord<T>> recordRDD) {
    // 对原始记录进行转化
    JavaPairRDD<String, HoodieRecord<T>> incomingRowKeyRecordPairRDD =
        recordRDD.mapToPair(record -> new Tuple2<>(record.getRecordKey(), record));
    // 对带有位置信息的记录也进行一次转化
    JavaPairRDD<String, Tuple2<HoodieRecordLocation, HoodieKey>> existingRecordKeyToRecordLocationHoodieKeyMap =
        keyLocationPairRDD.mapToPair(p -> new Tuple2<>(p._1.getRecordKey(), new Tuple2<>(p._2, p._1)));

    // 以recordKey进行一次左外连接，确定位置信息
    return incomingRowKeyRecordPairRDD.leftOuterJoin(existingRecordKeyToRecordLocationHoodieKeyMap).values().map(record -> {
      final HoodieRecord<T> hoodieRecord = record._1;
      final Optional<Tuple2<HoodieRecordLocation, HoodieKey>> recordLocationHoodieKeyPair = record._2;
      if (recordLocationHoodieKeyPair.isPresent()) {
        // Record key matched to file
        return getTaggedRecord(new HoodieRecord<>(recordLocationHoodieKeyPair.get()._2, hoodieRecord.getData()), Option.ofNullable(recordLocationHoodieKeyPair.get()._1));
      } else {
        return getTaggedRecord(hoodieRecord, Option.empty());
      }
    });
  }

其处理逻辑与父类处理逻辑相同，也是使用一次左外连接将位置信息推回至原始记录。

3. 总结

对于 HoodieGlobalBloomIndex而言，其是全局的索引，即会在所有分区内查找指定的recordKey，而非像 HoodieBloomIndex只在指定的分区内查找，同时在加载分区下所有最新文件时，其会首先获取所有分区，然后再获取所有分区下的最新文件，而非使用从原始记录中解析出来的分区路径。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-02-10，如有侵权请联系 cloudcommunity@tencent.com 删除

explode