Bloom Filter在Hudi中的應用

介紹

Bloom Filter可以用於檢索一個元素是否在一個集合中。它的優點是空間效率和查詢時間都遠遠超過一般的演算法，主要缺點是存在一定的誤判率：當其判斷元素存在時，實際上元素可能並不存在。而當判定不存在時，則元素一定不存在，Bloom Filter在對精確度要求不太嚴格的大資料量場景下運用十分廣泛。

引入

為何要引入Bloom Filter？這是Hudi為加快資料upsert採用的一種解決方案，即判斷record是否已經在檔案中存在，若存在，則更新，若不存在，則插入。對於upsert顯然無法容忍出現誤判，否則可能會出現應該插入和變成了更新的錯誤，那麼Hudi是如何解決誤判問題的呢？一種簡單辦法是當Bloom Filter判斷該元素存在時，再去檔案裡二次確認該元素是否真的存在；而當Bloom Filter判斷該元素不存在時，則無需讀檔案，通過二次確認的方法來規避Bloom Filter的誤判問題，實際上這也是Hudi採取的方案，值得一提的是，現在Delta暫時還不支援Bloom Filter，其判斷一條記錄是否存在是直接通過一次全表join來實現，效率比較低下。接下來我們來分析Bloom Filter在Hudi中的應用。

流程

Hudi從上游系統（Kafka、DFS等）消費一批資料後，會根據使用者配置的寫入模式（insert、upsert、bulkinsert）寫入Hudi資料集。而當配置為upsert時，意味著需要將資料插入更新至Hudi資料集，而第一步是需要標記哪些記錄已經存在，哪些記錄不存在，然後，對於存在的記錄進行更新，不存在記錄進行插入。

在HoodieWriteClient中提供了對應三種寫入模式的方法（#insert、#upsert、#bulkinsert），對於使用了Bloom Filter的#upsert方法而言，其核心原始碼如下

public JavaRDD<WriteStatus> upsert(JavaRDD<HoodieRecord<T>> records, final String commitTime) {
    ...
    // perform index loop up to get existing location of records
    JavaRDD<HoodieRecord<T>> taggedRecords = index.tagLocation(dedupedRecords, jsc, table);
    ...
    return upsertRecordsInternal(taggedRecords, commitTime, table, true);
}
 

可以看到首先利用索引給記錄打標籤，然後再進行更新，下面主要分析打標籤的過程。

對於索引，Hudi提供了四種索引方式的實現：HBaseIndex、HoodieBloomIndex、HoodieGlobalBloomIndex、InMemoryHashIndex，預設使用HoodieBloomIndex。其中HoodieGlobalBloomIndex與HoodieBloomIndex的區別是前者會讀取所有分割槽檔案，而後者只讀取記錄所存在的分割槽下的檔案。下面以HoodieBloomIndex為例進行分析。

HoodieBloomIndex#tagLocation核心程式碼如下

public JavaRDD<HoodieRecord<T>> tagLocation(JavaRDD<HoodieRecord<T>> recordRDD, JavaSparkContext jsc,
      HoodieTable<T> hoodieTable) {

    // Step 0: cache the input record RDD
    if (config.getBloomIndexUseCaching()) {
      recordRDD.persist(config.getBloomIndexInputStorageLevel());
    }

    // Step 1: Extract out thinner JavaPairRDD of (partitionPath, recordKey)
    JavaPairRDD<String, String> partitionRecordKeyPairRDD =
        recordRDD.mapToPair(record -> new Tuple2<>(record.getPartitionPath(), record.getRecordKey()));

    // Lookup indexes for all the partition/recordkey pair
    JavaPairRDD<HoodieKey, HoodieRecordLocation> keyFilenamePairRDD =
        lookupIndex(partitionRecordKeyPairRDD, jsc, hoodieTable);

    // Cache the result, for subsequent stages.
    if (config.getBloomIndexUseCaching()) {
      keyFilenamePairRDD.persist(StorageLevel.MEMORY_AND_DISK_SER());
    }

    // Step 4: Tag the incoming records, as inserts or updates, by joining with existing record keys
    // Cost: 4 sec.
    JavaRDD<HoodieRecord<T>> taggedRecordRDD = tagLocationBacktoRecords(keyFilenamePairRDD, recordRDD);

    if (config.getBloomIndexUseCaching()) {
      recordRDD.unpersist(); // unpersist the input Record RDD
      keyFilenamePairRDD.unpersist();
    }

    return taggedRecordRDD;
  }
 

該過程會快取記錄以便優化資料的載入。首先從記錄中解析出對應的分割槽路徑 -> key，接著檢視索引，然後將位置資訊（存在於哪個檔案）回推到記錄中。

HoodieBloomIndex#lookup核心程式碼如下

private JavaPairRDD<HoodieKey, HoodieRecordLocation> lookupIndex(
      JavaPairRDD<String, String> partitionRecordKeyPairRDD, final JavaSparkContext jsc,
      final HoodieTable hoodieTable) {
    // Obtain records per partition, in the incoming records
    Map<String, Long> recordsPerPartition = partitionRecordKeyPairRDD.countByKey();
    List<String> affectedPartitionPathList = new ArrayList<>(recordsPerPartition.keySet());

    // Step 2: Load all involved files as <Partition, filename> pairs
    List<Tuple2<String, BloomIndexFileInfo>> fileInfoList =
        loadInvolvedFiles(affectedPartitionPathList, jsc, hoodieTable);
    final Map<String, List<BloomIndexFileInfo>> partitionToFileInfo =
        fileInfoList.stream().collect(groupingBy(Tuple2::_1, mapping(Tuple2::_2, toList())));

    // Step 3: Obtain a RDD, for each incoming record, that already exists, with the file id,
    // that contains it.
    Map<String, Long> comparisonsPerFileGroup =
        computeComparisonsPerFileGroup(recordsPerPartition, partitionToFileInfo, partitionRecordKeyPairRDD);
    int safeParallelism = computeSafeParallelism(recordsPerPartition, comparisonsPerFileGroup);
    int joinParallelism = determineParallelism(partitionRecordKeyPairRDD.partitions().size(), safeParallelism);
    return findMatchingFilesForRecordKeys(partitionToFileInfo, partitionRecordKeyPairRDD, joinParallelism, hoodieTable,
        comparisonsPerFileGroup);
  }

該方法首先會計算出每個分割槽有多少條記錄和影響的分割槽有哪些，然後載入影響的分割槽的檔案，最後計算並行度後，開始找記錄真正存在的檔案。

對於#loadInvolvedFiles方法而言，其會查詢指定分割槽分割槽下所有的資料檔案（parquet格式），並且如果開啟了hoodie.bloom.index.prune.by.ranges，還會讀取檔案中的最小key和最大key（為加速後續的查詢）。

HoodieBloomIndex#findMatchingFilesForRecordKeys核心程式碼如下

JavaPairRDD<HoodieKey, HoodieRecordLocation> findMatchingFilesForRecordKeys(
      final Map<String, List<BloomIndexFileInfo>> partitionToFileIndexInfo,
      JavaPairRDD<String, String> partitionRecordKeyPairRDD, int shuffleParallelism, HoodieTable hoodieTable,
      Map<String, Long> fileGroupToComparisons) {
    JavaRDD<Tuple2<String, HoodieKey>> fileComparisonsRDD =
        explodeRecordRDDWithFileComparisons(partitionToFileIndexInfo, partitionRecordKeyPairRDD);

    if (config.useBloomIndexBucketizedChecking()) {
      Partitioner partitioner = new BucketizedBloomCheckPartitioner(shuffleParallelism, fileGroupToComparisons,
          config.getBloomIndexKeysPerBucket());

      fileComparisonsRDD = fileComparisonsRDD.mapToPair(t -> new Tuple2<>(Pair.of(t._1, t._2.getRecordKey()), t))
          .repartitionAndSortWithinPartitions(partitioner).map(Tuple2::_2);
    } else {
      fileComparisonsRDD = fileComparisonsRDD.sortBy(Tuple2::_1, true, shuffleParallelism);
    }

    return fileComparisonsRDD.mapPartitionsWithIndex(new HoodieBloomIndexCheckFunction(hoodieTable, config), true)
        .flatMap(List::iterator).filter(lr -> lr.getMatchingRecordKeys().size() > 0)
        .flatMapToPair(lookupResult -> lookupResult.getMatchingRecordKeys().stream()
            .map(recordKey -> new Tuple2<>(new HoodieKey(recordKey, lookupResult.getPartitionPath()),
                new HoodieRecordLocation(lookupResult.getBaseInstantTime(), lookupResult.getFileId())))
            .collect(Collectors.toList()).iterator());
  }

該方法首先會查詢記錄需要進行比對的檔案，然後再查詢的記錄的位置資訊。

其中，對於#explodeRecordRDDWithFileComparisons方法而言，其會藉助樹/連結串列結構構造的檔案過濾器來加速記錄對應檔案的查詢（每個record可能會對應多個檔案）。

而使用Bloom Filter的核心邏輯承載在HoodieBloomIndexCheckFunction，HoodieBloomIndexCheckFunction$LazyKeyCheckIterator該迭代器完成了記錄對應檔案的實際查詢過程，查詢的核心邏輯在computeNext`中，其核心程式碼如下

protected List<HoodieKeyLookupHandle.KeyLookupResult> computeNext() {

      List<HoodieKeyLookupHandle.KeyLookupResult> ret = new ArrayList<>();
      try {
        // process one file in each go.
        while (inputItr.hasNext()) {
          Tuple2<String, HoodieKey> currentTuple = inputItr.next();
          String fileId = currentTuple._1;
          String partitionPath = currentTuple._2.getPartitionPath();
          String recordKey = currentTuple._2.getRecordKey();
          Pair<String, String> partitionPathFilePair = Pair.of(partitionPath, fileId);

          // lazily init state
          if (keyLookupHandle == null) {
            keyLookupHandle = new HoodieKeyLookupHandle(config, hoodieTable, partitionPathFilePair);
          }

          // if continue on current file
          if (keyLookupHandle.getPartitionPathFilePair().equals(partitionPathFilePair)) {
            keyLookupHandle.addKey(recordKey);
          } else {
            // do the actual checking of file & break out
            ret.add(keyLookupHandle.getLookupResult());
            keyLookupHandle = new HoodieKeyLookupHandle(config, hoodieTable, partitionPathFilePair);
            keyLookupHandle.addKey(recordKey);
            break;
          }
        }

        // handle case, where we ran out of input, close pending work, update return val
        if (!inputItr.hasNext()) {
          ret.add(keyLookupHandle.getLookupResult());
        }
      } catch (Throwable e) {
        if (e instanceof HoodieException) {
          throw e;
        }
        throw new HoodieIndexException("Error checking bloom filter index. ", e);
      }

      return ret;
    }

該方法每次迭代只會處理一個檔案，每次處理時都會生成HoodieKeyLookupHandle，然後會新增recordKey，處理完後再獲取查詢結果。

其中HoodieKeyLookupHandle#addKey方法核心程式碼如下

public void addKey(String recordKey) {
    // check record key against bloom filter of current file & add to possible keys if needed
    if (bloomFilter.mightContain(recordKey)) {
      ...
      candidateRecordKeys.add(recordKey);
    }
    totalKeysChecked++;
  }

可以看到，這裡使用到了Bloom Filter來判斷該記錄是否存在，如果存在，則加入到候選佇列中，等待進一步判斷；若不存在，則無需額外處理，其中Bloom Filter會在建立HoodieKeyLookupHandle例項時初始化（從指定檔案中讀取Bloom Filter）。

HoodieKeyLookupHandle#getLookupResult方法核心程式碼如下

public KeyLookupResult getLookupResult() {
    ...
    HoodieDataFile dataFile = getLatestDataFile();
    List<String> matchingKeys =
        checkCandidatesAgainstFile(hoodieTable.getHadoopConf(), candidateRecordKeys, new Path(dataFile.getPath()));
    ...
    return new KeyLookupResult(partitionPathFilePair.getRight(), partitionPathFilePair.getLeft(),
        dataFile.getCommitTime(), matchingKeys);
  }

該方法首先獲取指定分割槽下的最新資料檔案，然後判斷資料檔案存在哪些recordKey，並將其封裝進KeyLookupResult後返回。其中#checkCandidatesAgainstFile會讀取檔案中所有的recordKey，判斷是否存在於candidateRecordKeys，這便完成了進一步確認。

到這裡即完成了record存在於哪些檔案的所有查詢，查詢完後會進行進一步處理，後續再給出分析。

總結

Hudi引入Bloom Filter是為了加速upsert過程，並將其存入parquet資料檔案中的Footer中，在讀取檔案時會從Footer中讀取該BloomFilter。在利用Bloom Filter來判斷記錄是否存在時，會採用二次確認的方式規避Bloom Filter的誤判問題。

相關推薦