Hudi 索引总结 - Parquet布隆过滤器写入过程

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住给大家分享一下。点击跳转到网站：https://www.captainai.net/dongkelun

前言

上篇文章提到 ：索引的逻辑主要是根据 parquet 文件中保存的索引信息，判断记录是否存在，如果不存在，代表是新增数据，如果记录存在则代表是更新数据，需要找到并设置 currentLocation。对于布隆索引来说，这里的索引信息其实是布隆过滤器，本篇文章主要是先总结布隆过滤器是如何保存到 parquet 文件中的（主要是源码调用逻辑）。

我们新写一个表的时候，最开始是没有parquet文件的，所以顺序应该是先将布隆过滤器写到 parquet 文件中，下次写数据的时候，先读取 parquet 文件中的布隆过滤器来验证表中是否存在该记录，有助于实现高效的更新和删除操作。

索引

索引是一个关键的步骤，它验证表中是否存在记录，并有助于实现高效的更新和删除操作。请注意本文中涵盖的索引是为写入端准备的，这与读取端索引不同。

对于有的索引类型是不涉及将索引信息保存到文件中的，比如 简单索引（simple index），而对于Bloom索引（Bloom Index）则需要将布隆过滤器以及最大值最小值等信息写到 parquet 文件中的元数据中。

Bloom索引（Bloom Index）

Bloom Index 最大限度地减少了用于查找的键和文件的数量，同时保持了较低的读取成本。

Bloom Index 采用 2 阶段过滤来减少用于查找的键和文件的数量。

  • 第一阶段涉及将输入键与使用存储在基本文件页脚中的最小和最大记录键值构建的间隔树进行比较。超出这些范围的键代表新插入，而其余键则被视为下一阶段的候选键。
  • 第二阶段根据反序列化的 Bloom 过滤器检查候选键，这有助于确定明确不存在的键和可能存在的键。然后使用筛选后的键和关联的基文件执行实际的文件查找，这些基文件随后返回用于标记的键和位置元组。

请注意，查找之前的过滤过程仅涉及读取文件页脚，因此读取成本较低。

源码分析

主要步骤包含布隆过滤器（BloomFilter）的创建、更新、和持久化。

fileWriter

HoodieCreateHandle 和 HoodieMergeHandle 中的 fileWriter 均为 HoodieParquetWriter 。
都是在构造函数中调用 HoodieFileWriterFactory.getFileWriter 创建

HoodieFileWriterFactory

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
public static <T extends HoodieRecordPayload, R extends IndexedRecord, I, K, O> HoodieFileWriter<R> getFileWriter(
    String instantTime, Path path, HoodieTable<T, I, K, O> hoodieTable, HoodieWriteConfig config, Schema schema,
    TaskContextSupplier taskContextSupplier) throws IOException {
  final String extension = FSUtils.getFileExtension(path.getName());
  if (PARQUET.getFileExtension().equals(extension)) {
    // config.populateMetaFields() 默认 true
    return newParquetFileWriter(instantTime, path, config, schema, hoodieTable, taskContextSupplier, config.populateMetaFields());
  }
  if (HFILE.getFileExtension().equals(extension)) {
    return newHFileFileWriter(instantTime, path, config, schema, hoodieTable, taskContextSupplier);
  }
  if (ORC.getFileExtension().equals(extension)) {
    return newOrcFileWriter(instantTime, path, config, schema, hoodieTable, taskContextSupplier);
  }
  throw new UnsupportedOperationException(extension + " format not supported yet.");
}

private static <T extends HoodieRecordPayload, R extends IndexedRecord> HoodieFileWriter<R> newParquetFileWriter(
    String instantTime, Path path, HoodieWriteConfig config, Schema schema, HoodieTable hoodieTable,
    TaskContextSupplier taskContextSupplier, boolean populateMetaFields) throws IOException {
  // enableBloomFilter 和 populateMetaFields 值一样，都为 true
  return newParquetFileWriter(instantTime, path, config, schema, hoodieTable, taskContextSupplier, populateMetaFields, populateMetaFields);
}

private static <T extends HoodieRecordPayload, R extends IndexedRecord> HoodieFileWriter<R> newParquetFileWriter(
    String instantTime, Path path, HoodieWriteConfig config, Schema schema, HoodieTable hoodieTable,
    TaskContextSupplier taskContextSupplier, boolean populateMetaFields, boolean enableBloomFilter) throws IOException {
  // 创建布隆过滤器
  Option<BloomFilter> filter = enableBloomFilter ? Option.of(createBloomFilter(config)) : Option.empty();
  HoodieAvroWriteSupport writeSupport = new HoodieAvroWriteSupport(new AvroSchemaConverter(hoodieTable.getHadoopConf()).convert(schema), schema, filter);

  HoodieAvroParquetConfig parquetConfig = new HoodieAvroParquetConfig(writeSupport, config.getParquetCompressionCodec(),
      config.getParquetBlockSize(), config.getParquetPageSize(), config.getParquetMaxFileSize(),
      hoodieTable.getHadoopConf(), config.getParquetCompressionRatio());

  return new HoodieParquetWriter<>(instantTime, path, parquetConfig, schema, taskContextSupplier, populateMetaFields);
}

populateMetaFields

启用时，填充所有元字段。禁用时，不会填充任何元字段，增量查询也不会起作用。这仅用于批处理的仅附加/不可变数据

简单说就是是否填充所有的元数据字段。
从上面的代码可以看到是否开启布隆过滤器和 populateMetaFields 的值一样。

1
2
3
4
5
6
7
8
9
10
public static final ConfigProperty<String> POPULATE_META_FIELDS = ConfigProperty
    .key("hoodie.populate.meta.fields")
    .defaultValue("true")
    .withDocumentation("When enabled, populates all meta fields. When disabled, no meta fields are populated "
        + "and incremental queries will not be functional. This is only meant to be used for append only/immutable data for batch processing");

public boolean populateMetaFields() {
  return Boolean.parseBoolean(getStringOrDefault(HoodieTableConfig.POPULATE_META_FIELDS,
      HoodieTableConfig.POPULATE_META_FIELDS.defaultValue()));
}

创建布隆过滤器

HoodieFileWriterFactory.createBloomFilter

1
2
3
4
5
private static BloomFilter createBloomFilter(HoodieWriteConfig config) {
  return BloomFilterFactory.createBloomFilter(config.getBloomFilterNumEntries(), config.getBloomFilterFPP(),
          config.getDynamicBloomFilterMaxNumEntries(),
          config.getBloomFilterType());
}

几个参数：

  • config.getBloomFilterNumEntries : BLOOM_FILTER_NUM_ENTRIES_VALUE hoodie.index.bloom.num_entries 默认值 6000
  • config.getBloomFilterFPP ：BLOOM_FILTER_FPP_VALUE hoodie.index.bloom.fpp 默认值 0.000000001
  • config.getDynamicBloomFilterMaxNumEntries ：BLOOM_INDEX_FILTER_DYNAMIC_MAX_ENTRIES hoodie.bloom.index.filter.dynamic.max.entries 默认值 100000
  • config.getBloomFilterType() ： BLOOM_FILTER_TYPE hoodie.bloom.index.filter.type 默认值 BloomFilterTypeCode.DYNAMIC_V0.name

BloomFilterFactory.createBloomFilter

1
2
3
4
5
6
7
8
9
10
11
public static BloomFilter createBloomFilter(int numEntries, double errorRate, int maxNumberOfEntries,
                                            String bloomFilterTypeCode) {
  if (bloomFilterTypeCode.equalsIgnoreCase(BloomFilterTypeCode.SIMPLE.name())) {
    return new SimpleBloomFilter(numEntries, errorRate, Hash.MURMUR_HASH);
  } else if (bloomFilterTypeCode.equalsIgnoreCase(BloomFilterTypeCode.DYNAMIC_V0.name())) {
    // 默认值 BloomFilterTypeCode.DYNAMIC_V0.name
    return new HoodieDynamicBoundedBloomFilter(numEntries, errorRate, Hash.MURMUR_HASH, maxNumberOfEntries);
  } else {
    throw new IllegalArgumentException("Bloom Filter type code not recognizable " + bloomFilterTypeCode);
  }
}

默认值 BloomFilterTypeCode.DYNAMIC_V0.name ，返回 HoodieDynamicBoundedBloomFilter

writeSupport

HoodieParquetWriter 中的 writeSupport 为 HoodieAvroWriteSupport

1
HoodieAvroWriteSupport writeSupport = new HoodieAvroWriteSupport(new AvroSchemaConverter(hoodieTable.getHadoopConf()).convert(schema), schema, filter);

更新布隆过滤器

通过 HoodieAvroWriteSupport.add 更新布隆过滤器，将 recordKey 添加到 BloomFilter 中，调用链：

  • Insert : HoodieCreateHandle.write -> HoodieParquetWriter.writeAvro/writeAvroWithMetadata -> HoodieAvroWriteSupport.add
  • Upsert : HoodieMergeHandle.write -> HoodieParquetWriter.writeAvro/writeAvroWithMetadata-> HoodieAvroWriteSupport.add 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public void add(String recordKey) {
  if (bloomFilterOpt.isPresent()) {
    // HoodieDynamicBoundedBloomFilter.add -> InternalDynamicBloomFilter.add -> org.apache.hadoop.util.bloom.BloomFilter
    this.bloomFilterOpt.get().add(recordKey);
    if (minRecordKey != null) {
      minRecordKey = minRecordKey.compareTo(recordKey) <= 0 ? minRecordKey : recordKey;
    } else {
      minRecordKey = recordKey;
    }

    if (maxRecordKey != null) {
      maxRecordKey = maxRecordKey.compareTo(recordKey) >= 0 ? maxRecordKey : recordKey;
    } else {
      maxRecordKey = recordKey;
    }
  }
}

这里主要调用 HoodieDynamicBoundedBloomFilter.add 继而调用 InternalDynamicBloomFilter.add 最终将 recordKey 添加到 org.apache.hadoop.util.bloom.BloomFilter 中,然后更新最大值和最小值。

写布隆过滤器

Insert 和 Update 最终都通过调用 HoodieParquetWriter.close 实现将布隆过滤写到 Parquet Metadata :

1
2
3
4
5
writer.model.name : avro
org.apache.hudi.bloomfilter : bloomfilter
hoodie_min_record_key : minRecordKey
hoodie_max_record_key : maxRecordKey
hoodie_bloom_filter_type_code : DYNAMIC_V0

调用链：

  • Insert : BoundedInMemoryQueueConsumer.consume -> BoundedInMemoryQueueConsumer.finish -> CopyOnWriteInsertHandler.finish -> CopyOnWriteInsertHandler.closeOpenHandles -> HoodieCreateHandle.close -> HoodieParquetWriter.close
    -> org.apache.parquet.hadoop.ParquetWriter.close -> org.apache.parquet.hadoop.InternalParquetRecordWriter.close -> org.apache.parquet.hadoop.ParquetFileWriter.end -> ParquetFileWriter.serializeFooter

  • Upsert : BaseJavaCommitActionExecutor.handleUpdate -> handleUpdateInternal -> JavaMergeHelper.runMerge -> HoodieMergeHandle.close -> HoodieParquetWriter.close
    -> org.apache.parquet.hadoop.ParquetWriter.close -> org.apache.parquet.hadoop.InternalParquetRecordWriter.close -> org.apache.parquet.hadoop.ParquetFileWriter.end -> ParquetFileWriter.serializeFooter

HoodieParquetWriter.close

1
2
3
4
5
6
7
8
9
10
public void close() throws IOException {
  try {
    writer.close();
  } catch (InterruptedException e) {
    throw new IOException(e);
  } finally {
    // release after the writer closes in case it is used for a last flush
    codecFactory.release();
  }
}

InternalParquetRecordWriter.close

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public void close() throws IOException, InterruptedException {
  if (!closed) {
    flushRowGroupToStore();
    // 这里调用 HoodieAvroWriteSupport.finalizeWrite 获取布隆过滤器、最大值、最小值等。
    FinalizedWriteContext finalWriteContext = writeSupport.finalizeWrite();
    Map<String, String> finalMetadata = new HashMap<String, String>(extraMetaData);
    // 返回 avro
    String modelName = writeSupport.getName();
    if (modelName != null) {
      // writer.model.name : avro
      finalMetadata.put(ParquetWriter.OBJECT_MODEL_NAME_PROP, modelName);
    }
    // 将 finalWriteContext.getExtraMetaData 添加到 finalMetadata 中
    finalMetadata.putAll(finalWriteContext.getExtraMetaData());
    // 调用 ParquetFileWriter.end 方法将 finalMetadata 写到 parquet 页脚中
    parquetFileWriter.end(finalMetadata);
    closed = true;
  }
}

HoodieAvroWriteSupport.finalizeWrite

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Override
public WriteSupport.FinalizedWriteContext finalizeWrite() {
  HashMap<String, String> extraMetaData = new HashMap<>();
  if (bloomFilterOpt.isPresent()) {
    // org.apache.hudi.bloomfilter : bloomfilter (序列化的)
    extraMetaData.put(HOODIE_AVRO_BLOOM_FILTER_METADATA_KEY, bloomFilterOpt.get().serializeToString());
    if (minRecordKey != null && maxRecordKey != null) {
      // hoodie_min_record_key : minRecordKey
      // hoodie_max_record_key : maxRecordKey
      extraMetaData.put(HOODIE_MIN_RECORD_KEY_FOOTER, minRecordKey);
      extraMetaData.put(HOODIE_MAX_RECORD_KEY_FOOTER, maxRecordKey);
    }
    // getBloomFilterTypeCode().name() : DYNAMIC_V0
    if (bloomFilterOpt.get().getBloomFilterTypeCode().name().contains(HoodieDynamicBoundedBloomFilter.TYPE_CODE_PREFIX)) {
      // hoodie_bloom_filter_type_code : DYNAMIC_V0
      extraMetaData.put(HOODIE_BLOOM_FILTER_TYPE_CODE, bloomFilterOpt.get().getBloomFilterTypeCode().name());
    }
  }
  // 将 extraMetaData 值赋给 WriteSupport.extraMetaData 并返回 FinalizedWriteContext
  return new WriteSupport.FinalizedWriteContext(extraMetaData);
}
}

writeSupport.getName -> HoodieAvroWriteSupport.getName -> AvroWriteSupport.getName

1
2
3
4
@Override
public String getName() {
  return "avro";
}

ParquetFileWriter.end

1
2
3
4
5
6
7
8
public void end(Map<String, String> extraMetaData) throws IOException {
  state = state.end();
  LOG.debug("{}: end", out.getPos());
  // 将 extraMetaData 赋给 FileMetaData.keyValueMetaData 
  this.footer = new ParquetMetadata(new FileMetaData(schema, extraMetaData, Version.FULL_VERSION), blocks);
  serializeFooter(footer, out);
  out.close();
}

ParquetFileWriter.serializeFooter

1
2
3
4
5
6
7
8
9
10
private static void serializeFooter(ParquetMetadata footer, PositionOutputStream out) throws IOException {
  long footerIndex = out.getPos();
  // 将 footer 转换为 org.apache.parquet.format.FileMetaData
  org.apache.parquet.format.FileMetaData parquetMetadata = metadataConverter.toParquetMetadata(CURRENT_VERSION, footer);
  // 调用 writeFileMetaData 将 parquetMetadata 写到 parquet 元数据中
  writeFileMetaData(parquetMetadata, out);
  LOG.debug("{}: footer length = {}" , out.getPos(), (out.getPos() - footerIndex));
  BytesUtils.writeIntLittleEndian(out, (int) (out.getPos() - footerIndex));
  out.write(MAGIC);
}

赏

感谢您的支持!

支付宝
微信

本文由 董可伦 发表于 伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。

非商业转载请注明作者及出处。商业转载请联系作者本人。

本文标题：Hudi 索引总结 - Parquet布隆过滤器写入过程

本文链接：https://dongkelun.com/2024/07/10/hudi/hudiIndex-wirte/

欢迎关注我的公众号

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

2016年本科毕业于合肥工业大学<br><br>信息与计算科学专业<br><br>目前在济南一家公司从事大数据的工作~~