0
点赞
收藏
分享

微信扫一扫

计算机网络面经-HTTP的8种请求方式

就是耍帅 02-25 15:30 阅读 3
1、背景

Hudi程序中upsert操作频繁,过多的删除和回滚操作,导致集群RPC持续偏高

2、描述

hudi采用的是mvcc设计,提供了清理工具cleaner来把旧版本的文件分片删除,默认开启了清理功能,可以防止文件系统的存储空间和文件数量的无限增长。

3、清理保留策略

清理旧文件需要考虑数据查询的情况,有些长查询会占用着旧版本的文件,需要设置合适的清理策略来保留一定数量的commit或者文件版本,以提高系统的容错性

  • KEEP_LATEST_COMMITS:默认策略,表示保留最后n次提交,默认为10,通过参数hoodie.cleaner.commits.retained​或clean.retain_commits​(flink)设置
  • KEEP_LATEST_FILE_VERSIONS:保留最后n个文件版本,默认为3,通过参数hoodie.cleaner.fileversions.retained设置​
  • KEEP_LATEST_BY_HOURS:保留最后n小时,默认24小时,通过参数hoodie.cleaner.hours.retained设置,这是0.11版本后新增的
4、 清理触发策略

目前仅支持一种触发清理的策略:CleaningTriggerStrategy#NUM_COMMITS,即根据提交的次数,默认为1,可以通过设置参数hoodie.clean.max.commits​进行修改,在flink job的每次checkpoint时都会进行触发策略的条件判断,所以在两次chekpoint之间发生过1次或n次提交,都会触发清理动作。

5、清理流程分析 
5.1、清理器初始化

清理逻辑是被包装成一个flink sink,在HoodieTableSink#getSinkRuntimeProvider中进行初始化

如果是mor表且开启了异步合并(compaction.async.enabled),则创建CompactionCommitSink,继承了CleanFunction,所以包含了清理逻辑,这是由于SQL API中一个SinkRuntimeProvider不支持多个sink.
否则,直接将CleanFunction作为sink,这种情况必需启用异步清理配置clean.async.enabled,因为CleanFunction的主要方法都判断了是否为异步清理。

5.2、清理启动入口
  1. compact成功后同步清理
    需要满足条件:1)mor表,2)启用异步合并compaction.async.enabled,3)禁用异步清理clean.async.enabled。入代码在CompactionCommitSink#doCommit中:
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
  this.writeClient.clean();
}
  1. checkpoint时异步清理
    需要满足条件:1)非mor表或启用异步合并compaction.async.enabled,2)启用异步清理clean.async.enabled。入口代码在CleanFunction#snapshotState中:
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
    this.writeClient.startAsyncCleaning();
    this.isCleaning = true;
}
 6、清理逻辑执行

清理逻辑的流程,主要包含有三个步骤:生成清理计划、刷新ActiveTimeline、执行清理计划

  1. 如果处理的instant状态为requested需要先转换为inflight状态(生成xxx.clean.inflight文件),表示开始清理。
  2. 执行清理clean(context, cleanerPlan),根据清理计划的数据进行文件删除即可,首先删除每个分区下需要清理的文件,然后删除需清理的分区目录,最后收集统计数据返回。
  3. 清理成功后将infight状态转换为completed状态,表示清理完成。

参考:

All Configurations | Apache Hudi

hudi系列-旧文件清理(clean)-天翼云开发者社区 - 天翼云 (ctyun.cn)

举报

相关推荐

0 条评论