Apache Hudi初探(五)(与flink的结合)--Flink 中hudi clean操作

这篇具有很好参考价值的文章主要介绍了Apache Hudi初探(五)(与flink的结合)--Flink 中hudi clean操作。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

背景

本文主要是具体说说Flink中的clean操作的实现

杂说闲谈

在flink中主要是CleanFunction函数:

  @Override
  public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
    this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
    String instantTime = HoodieActiveTimeline.createNewInstantTime();
    LOG.info(String.format("exec clean with instant time %s...", instantTime));
    executor.execute(() -> writeClient.clean(instantTime), "wait for cleaning finish");
  }

  @Override
  public void notifyCheckpointComplete(long l) throws Exception {
    if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && isCleaning) {
      executor.execute(() -> {
        try {
          this.writeClient.waitForCleaningFinish();
        } finally {
          // ensure to switch the isCleaning flag
          this.isCleaning = false;
        }
      }, "wait for cleaning finish");
    }
  }

  @Override
  public void snapshotState(FunctionSnapshotContext context) throws Exception {
    if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
      try {
        this.writeClient.startAsyncCleaning();
        this.isCleaning = true;
      } catch (Throwable throwable) {
        // catch the exception to not affect the normal checkpointing
        LOG.warn("Error while start async cleaning", throwable);
      }
    }
  }

  • open函数

    • writeClient =FlinkWriteClients.createWriteClient(conf, getRuntimeContext())
      创建FlinkWriteClient,用于写hudi数据

    • this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
      创建一个只有一个线程的线程池,改线程池的主要作用来异步执行hudi写操作

    • executor.execute(() -> writeClient.clean(instantTime)
      异步执行hudi的清理操作,该clean函数的主要代码如下:

         if (!tableServicesEnabled(config)) {
           return null;
         }
         final Timer.Context timerContext = metrics.getCleanCtx();
         CleanerUtils.rollbackFailedWrites(config.getFailedWritesCleanPolicy(),
             HoodieTimeline.CLEAN_ACTION, () -> rollbackFailedWrites(skipLocking));
      
         HoodieTable table = createTable(config, hadoopConf);
         if (config.allowMultipleCleans() || !table.getActiveTimeline().getCleanerTimeline().filterInflightsAndRequested().firstInstant().isPresent()) {
           LOG.info("Cleaner started");
           // proceed only if multiple clean schedules are enabled or if there are no pending cleans.
           if (scheduleInline) {
             scheduleTableServiceInternal(cleanInstantTime, Option.empty(), TableServiceType.CLEAN);
             table.getMetaClient().reloadActiveTimeline();
           }
         }
      
         // Proceeds to execute any requested or inflight clean instances in the timeline
         HoodieCleanMetadata metadata = table.clean(context, cleanInstantTime, skipLocking);
         if (timerContext != null && metadata != null) {
           long durationMs = metrics.getDurationInMs(timerContext.stop());
           metrics.updateCleanMetrics(durationMs, metadata.getTotalFilesDeleted());
           LOG.info("Cleaned " + metadata.getTotalFilesDeleted() + " files"
               + " Earliest Retained Instant :" + metadata.getEarliestCommitToRetain()
               + " cleanerElapsedMs" + durationMs);
         }
         return metadata;
      
      • CleanerUtils.rollbackFailedWrites(config.getFailedWritesCleanPolicy(),HoodieTimeline.CLEAN_ACTION,() -> rollbackFailedWrites *
        根据配置
        hoodie.cleaner.policy.failed.writes* 默认是EAGER,也就是在写数据失败的时候,会立即进行这次写失败的数据的清理,在这种情况下,
        就不会执行rollbackFailedWrites操作,也就是回滚写失败文件的操作

      • HoodieTable table = createTable *
        创建
        HoodieFlinkMergeOnReadTable*类型的hudi表,用来做clean等操作

      • scheduleTableServiceInternal
        如果hoodie.clean.allow.multiple为true(默认为true)或者没有正在运行中clean操作,则会生成Clean计划
        这里最终调用的是FlinkWriteClient.scheduleCleaning方法,即CleanPlanActionExecutor.execute方法

        这里最重要的就是requestClean方法:

        CleanPlanner<T, I, K, O> planner = new CleanPlanner<>(context, table, config);
        Option<HoodieInstant> earliestInstant = planner.getEarliestCommitToRetain();
        List<String> partitionsToClean = planner.getPartitionPathsToClean(earliestInstant)    
        int cleanerParallelism = Math.min(partitionsToClean.size(), config.getCleanerParallelism());
        Map<String, Pair<Boolean, List<CleanFileInfo>>> cleanOpsWithPartitionMeta = context
            .map(partitionsToClean, partitionPathToClean -> Pair.of(partitionPathToClean, planner.getDeletePaths(partitionPathToClean)), cleanerParallelism)
            .stream()
            .collect(Collectors.toMap(Pair::getKey, Pair::getValue))    
        Map<String, List<HoodieCleanFileInfo>> cleanOps = cleanOpsWithPartitionMeta.entrySet().stream()
            .collect(Collectors.toMap(Map.Entry::getKey,
                e -> CleanerUtils.convertToHoodieCleanFileInfoList(e.getValue().getValue())))    
        List<String> partitionsToDelete = cleanOpsWithPartitionMeta.entrySet().stream().filter(entry -> entry.getValue().getKey()).map(Map.Entry::getKey)
            .collect(Collectors.toList())    
        return new HoodieCleanerPlan(earliestInstant
            .map(x -> new HoodieActionInstant(x.getTimestamp(), x.getAction(), x.getState().name())).orElse(null),
            planner.getLastCompletedCommitTimestamp(),
            config.getCleanerPolicy().name(), CollectionUtils.createImmutableMap(),
            CleanPlanner.LATEST_CLEAN_PLAN_VERSION, cleanOps, partitionsToDelete)    
        
        • planner.getEarliestCommitToRetain();
          根据保留策略,获取到最早需要保留的commit的HoodieInstant,在这里会兼顾考虑到hoodie.cleaner.commits.retained(默认是10)以及hoodie.cleaner.hours.retained默认是24小时以及hoodie.cleaner.policy策略(默认是KEEP_LATEST_COMMITS)
        • planner.getPartitionPathsToClean(earliestInstant);
          根据保留的最新commit的HoodieInstant,得到要删除的分区,这里会根据配置hoodie.cleaner.incremental.mode(默认是true)来进行增量清理,
          这个时候就会根据上一次已经clean的信息,只需要删除差量的分区数据就行
        • cleanOpsWithPartitionMeta = context
          根据上面得到的需要删除的分区信息,获取需要删除的文件信息,具体的实现可以参考CleanPlanner.getFilesToCleanKeepingLatestCommits
          这里的操作主要是先通过fileSystemView获取分区下所有的FileGroup,之后再获取每个FileGroup下的所有的FileSlice(这里的FileSlice就有版本的概念,也就是commit的版本),之后再与最新保留的commit的时间戳进行比较得到需要删除的文件信息
        • new HoodieCleanerPlan
          最后组装成HoodieCleanPlan的计划,并且在外层调用table.getActiveTimeline().saveToCleanRequested(cleanInstant, TimelineMetadataUtils.serializeCleanerPlan(cleanerPlan)); 方法把clean request的状态存储到对应的.hoodie目录下,并建立一个xxxx.clean.requested的元数据文件
      • table.getMetaClient().reloadActiveTimeline()
        重新加载timeline,便于过滤出来刚才scheduleTableServiceInternal操作生成的xxxxxxxxxxxxxx.clean.requested的元数据文件

      • table.clean(context, cleanInstantTime, skipLocking)
        真正执行clean的部分,主要是调用CleanActionExecutor.execute的方法,最终调用的是*runPendingClean(table, hoodieInstant)*方法:

         HoodieCleanerPlan cleanerPlan = CleanerUtils.getCleanerPlan(table.getMetaClient(), cleanInstant);
         return runClean(table, cleanInstant, cleanerPlan);
        

        首先是反序列化CleanPlan,然后在进行清理,主要是删除1. 如果没有满足的分区,直接删除该分区,2. 否则删除该分区下的满足条件的文件,最后返回HoodieCleanStat包含删除的文件信息等。

  • snapshotState方法

    • 如果clean.async.enabled是true(默认是true),并且不是正在进行clean动作,则会进行异步清理
      this.writeClient.startAsyncCleaning(); 这里最终也是调用的writeClient.clean方法。
    • this.isCleaning = true;
      设置标志位,用来保证clean操作的有序性
  • notifyCheckpointComplete方法文章来源地址https://www.toymoban.com/news/detail-736703.html

    • 如果clean.async.enabled是true(默认是true),并且正在进行clean动作,则等待clean操作完成,
      并且设置清理标识位,用来和snapshotState方法进行呼应以保证clean操作的有序性

到了这里,关于Apache Hudi初探(五)(与flink的结合)--Flink 中hudi clean操作的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Apache Hudi初探(九)(与spark的结合)--非bulk_insert模式

    之前讨论的都是’hoodie.datasource.write.operation’:\\\'bulk_insert’的前提下,在这种模式下,是没有json文件的已形成如下的文件: 因为是 bulk insert 操作,所以没有去重的需要,所以直接采用spark原生的方式, 以下我们讨论非spark原生的方式, 继续Apache Hudi初探(八)(与spark的结合)–非

    2024年02月08日
    浏览(37)
  • 实时数据湖 Flink Hudi 实践探索

    导读: 首先做个自我介绍,我目前在阿里云云计算平台,从事研究 Flink 和 Hudi 结合方向的相关工作。 目前,Flink + Hudi 的方案推广大概已经有了一年半的时间,在国内流行度也已比较高,主流的公司也会尝试去迭代他们的数仓方案。所以,今天我介绍的主题是 Flink 和 Hudi 在

    2024年01月16日
    浏览(51)
  • 【数据湖Hudi-10-Hudi集成Flink-读取方式&限流&写入方式&写入模式&Bucket索引】

    当前表默认是快照读取,即读取最新的全量快照数据并一次性返回。通过参数 read.streaming.enabled 参数开启流读模式,通过 read.start-commit 参数指定起始消费位置,支持指定 earliest 从最早消费。 1.with参数 名称 Required 默认值 说明 read.streaming.enabled false false 设置 true 开启流读模式

    2024年02月14日
    浏览(46)
  • Flink Catalog 解读与同步 Hudi 表元数据的最佳实践

    博主历时三年精心创作的《大数据平台架构与原型实现:数据中台建设实战》一书现已由知名IT图书品牌电子工业出版社博文视点出版发行,点击《重磅推荐:建大数据平台太难了!给我发个工程原型吧!》了解图书详情,京东购书链接:https://item.jd.com/12677623.html,扫描左侧

    2024年02月22日
    浏览(46)
  • Hudi(16):Hudi集成Flink之读取方式

    目录 0. 相关文章链接 1. 流读(Streaming Query) 2. 增量读取(Incremental Query) 3. 限流  Hudi文章汇总          当前表默认是快照读取,即读取最新的全量快照数据并一次性返回。通过参数read.streaming.enabled 参数开启流读模式,通过 read.start-commit 参数指定起始消费位置,支

    2024年02月06日
    浏览(67)
  • Hudi(17):Hudi集成Flink之写入方式

    目录 0. 相关文章链接 1. CDC 数据同步 1.1. 准备MySQL表 1.2. flink读取mysql binlog并写入kafka 1.3. flink读取kafka数据并写入hudi数据湖 1.4. 使用datafaker插入数据 1.5. 统计数据入Hudi情况 1.6. 实时查看数据入湖情况 2. 离线批量导入 2.1. 原理 2.2. WITH 参数 2.3. 案例 3. 全量接增量 3.1. 

    2024年02月05日
    浏览(40)
  • Hudi(19):Hudi集成Flink之索引和Catalog

    目录 0. 相关文章链接 1. Bucket索引(从 0.11 开始支持) 1.1. WITH参数 1.2. 和 state 索引的对比 2. Hudi Catalog(从 0.12.0 开始支持) 2.1. 概述 2.2. WITH 参数 2.3. 使用dfs方式  Hudi文章汇总          默认的 flink 流式写入使用 state 存储索引信息:primary key 到 fileId 的映射关系。当

    2024年02月05日
    浏览(39)
  • 大数据Hadoop之——Apache Hudi 数据湖实战操作(FlinkCDC)

    Hudi(Hadoop Upserts Deletes and Incrementals) ,简称 Hudi ,是一个 流式数据湖平台 ,关于Hudi的更多介绍可以参考我以下几篇文章: 大数据Hadoop之——新一代流式数据湖平台 Apache Hudi 大数据Hadoop之——Apache Hudi 数据湖实战操作(Spark,Flink与Hudi整合) 这里主要讲解Hive、Trino、Starr

    2023年04月20日
    浏览(40)
  • Hudi集成Flink

    安装Maven 1)上传apache-maven-3.6.3-bin.tar.gz到/opt/software目录,并解压更名 tar -zxvf apache-maven-3.6. 3 -bin.tar.gz -C /opt/module/ mv   apache -maven-3.6. 3  maven 2)添加环境变量到/etc/profile中 sudo  vim /etc/profile #MAVEN_HOME export MAVEN_HOME=/opt/module/maven export PATH=$PATH:$MAVEN_HOME/bin 3)测试安装结果 sourc

    2023年04月13日
    浏览(34)
  • Hudi(四)集成Flink(2)

            当前表 默认是快照读取 ,即读取最新的全量快照数据并一次性返回。通过参数 read.streaming.enabled 参数开启流读模式,通过 read.start-commit 参数指定起始消费位置,支持指定 earliest 从最早消费。 1、WITH参数 名称 Required 默认值 说明 read.streaming.enabled false false 设置

    2024年02月07日
    浏览(37)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包