Hudi 源码 | 索引总结 - tag/tagLocation

2024-07-15

前言

接上篇文章和之前的总结的源码文章，本文总结源码 tag/tagLocation ，对应功能：根据索引信息判断记录是否存在，如果不存在，代表是新增数据，如果记录存在则代表是更新数据，需要找到并设置 currentLocation。

tag

AbstractWriteHelper.tag

private I tag(
    I dedupedRecords, HoodieEngineContext context, HoodieTable<T, I, K, O> table) {
  // perform index loop up to get existing location of records
  // 执行索引循环以获取记录的现有位置
  // 对于 Java Client 这里 table 为 HoodieJavaCopyOnWriteTable
  return table.getIndex().tagLocation(dedupedRecords, context, table);
}

table.getIndex()

对于 Java Client 这里 table 为 HoodieJavaCopyOnWriteTable ， HoodieJavaCopyOnWriteTable.getIndex() -> HoodieJavaTable.getIndex

protected HoodieIndex<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> getIndex(HoodieWriteConfig config, HoodieEngineContext context) {
  return JavaHoodieIndex.createIndex(config);
}

public static HoodieIndex createIndex(HoodieWriteConfig config) {
  // first use index class config to create index.
  if (!StringUtils.isNullOrEmpty(config.getIndexClass())) {
    Object instance = ReflectionUtils.loadClass(config.getIndexClass(), config);
    if (!(instance instanceof HoodieIndex)) {
      throw new HoodieIndexException(config.getIndexClass() + " is not a subclass of HoodieIndex");
    }
    return (JavaHoodieIndex) instance;
  }

  // TODO more indexes to be added
  // 从这里看出，当前版本（0.9.0），Java Client 只支持两种索引类型：INMEMORY 和 BLOOM
  switch (config.getIndexType()) {
    case INMEMORY:
      return new JavaInMemoryHashIndex(config);
    case BLOOM:
      return new JavaHoodieBloomIndex(config);
    default:
      throw new HoodieIndexException("Unsupported index type " + config.getIndexType());
  }
}

因为指定了索引类型为 BLOOM , 所以这里返回 JavaHoodieBloomIndex 。

more >>

展开全文 >>

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

2024-07-10

前言

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

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

索引

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

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

more >>

展开全文 >>

Hudi 写入流程（图）

2024-07-04

前言

主要为之前总结的源码文章补充流程图。
总结一下整体流程

说明

之前以Java Client为例,总结了 Insert 源码的整体流程及部分源码，由于各种原因，没有总结完。长时间不看这方面的源码，容易忘记，之前没有总结流程图，现在回忆起来比较麻烦，不如看流程图方便快捷。所以先补充总结一下之前文章中的流程图，再继续学习总结后面的源码，这样比较方便。

流程图

主要分为自己总结以及借助Shiyan Xu总结的博客中的流程图一起理解。

more >>

展开全文 >>

Flink 重启策略和故障恢复策略

2024-06-24

前言

主要总结 Flink 重启策略

官方文档

版本

Flink 1.15.4

Task 故障恢复

Task Failure Recovery
当 Task 发生故障时，Flink 需要重启出错的 Task 以及其他受到影响的 Task ，以使得作业恢复到正常执行状态。
Flink 通过重启策略和故障恢复策略来控制 Task 重启：重启策略决定是否可以重启以及重启的间隔；故障恢复策略决定哪些 Task 需要重启。

more >>

展开全文 >>

Hudi extraMetadata 研究总结

2024-06-13

前言

研究总结 Hudi extraMetadata ，记录研究过程。主要目的是通过 extraMetadata 保存 source 表的 commitTime （checkpoint），来实现增量读Hudi表写Hudi表时，保存增量读状态的事务性，实现类似于流任务中的 exactly-once

背景需求

有个需求：增量读Hudi表关联其他Hudi表然后做一些复杂的业务处理逻辑,然后将结果表保存到目标Hudi表中

然后就有问题：如何保证保存目标Hudi表和保存 endCommiTime 是事务的。具体如下
1、Spark支持增量读Hudi表，需要传入起（始） commitTime。
2、上一批数据处理完成后应该保存上次增量读的的 endCommiTime,作为下一批增量读的的 beginCommitTime
3、可以选择将上一批的 endCommiTime 保存到 HDFS 或者数据库表中，但是不能保证事务。（并且还需要借助外部表或者存储目录，如果能通过Hudi本身解决，就比较完美了）
4、也就是可能存在保存目标Hudi表成功，但是保存 endCommiTime 失败的情况

这时就想到了 DeltaStreamer 中是通过设置 checkpoint 来实现的，具体是通过将 commitTime 的值保存到目标表 .commit 元数据文件中(extraMetadata->deltastreamer.checkpoint.key)。我们可以参考 DeltaStreamer 中的实现逻辑，来实现我们的需求。

more >>

展开全文 >>