Flink BucketingSink 源码分析

2024-06-02 17:58

本文主要是介绍Flink BucketingSink 源码分析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

 

 

0x1 摘要

BucketingSink类提供了非常完美的功能支持数据落HDFS,在实际业务中不建议自己去实现,直接采用此类可以避免一些坑。注:此文基于Flink 1.6.3 版本源码。

0x2 BucketingSink 类结构分析


我们关注RichSinkFunction、CheckpointedFunction、CheckpointListener三个父类

0x3 先看使用例子

BucketingSink<Object> sink = new BucketingSink<>(path);

sink.setBucketer(new DateTimeBucketer<>("yyyy/MM/dd"));

// 字符串形式输出

sink.setWriter(new StringWriter<>());

// 每个文件最大小限制256M,达到后关闭或创建新文件

sink.setBatchSize(1024 * 1024 * 256L);

// 设定批次滚动时间翻滚间隔30分钟,达到后关闭或创建新文件,和上面的`batchSize`双重检查决定

sink.setBatchRolloverInterval(30 * 60 * 1000L);

 

// 设定不活动桶时间阈值,超过此值便关闭文件

sink.setInactiveBucketThreshold(3 * 60 * 1000L);

// 设定检查不活动桶的频率

sink.setInactiveBucketCheckInterval(30 * 1000L);

 

// 设置正在写入的文件后缀,和默认后缀一致

sink.setInProgressSuffix(".in-progress");

// 一旦part文件关闭写入,变为挂起状态,和默认后缀一致。

// 注意:只有checkpoint成功后,.pending文件才会转为已完成状态。如果checkpoint不成功,.pending文件永不转变为完成状态。

sink.setPendingSuffix(".pending");

0x4 数据写入

我们先想一下数据流进来后如何写到HDFS文件中?最开始我的想法很简单,通过FileSystem创建一个文件流直接写入就行。那我们再往深一点想,写入发生异常了怎么办?写入异常后数据怎么恢复?怎么确定数据一致性?以上问题BucketingSink都已经帮你处理好。
下面从RichSinkFunction类的invoke方法开始一步步分析源码:

public void invoke(T value) throws Exception {

 // 通过分桶策略来初始化路径,使用例子中指定DateTimeBucketer策略,具体分桶实现看getBucketPath源码

 Path bucketPath = bucketer.getBucketPath(clock, new Path(basePath), value);

 

 long currentProcessingTime = processingTimeService.getCurrentProcessingTime();

 

 // 初始化桶状态

 BucketState<T> bucketState = state.getBucketState(bucketPath);

 if (bucketState == null) {

  bucketState = new BucketState<>(currentProcessingTime);

  state.addBucketState(bucketPath, bucketState);

 }

 

 // 判断是否需要滚动文件,下面详细介绍 shouldRoll 方法

 if (shouldRoll(bucketState, currentProcessingTime)) {

  openNewPartFile(bucketPath, bucketState);

 }

 

 // 写入数据

 bucketState.writer.write(value);

   

 //记录最近一次写入时间,按时间策略滚动有用

 bucketState.lastWrittenToTime = currentProcessingTime;

}

shouldRoll方法源码:

private boolean shouldRoll(BucketState<T> bucketState, long currentProcessingTime) throws IOException {

 boolean shouldRoll = false;

 int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();

   

 //bucketState初始状态时,设置为需要滚动

 if (!bucketState.isWriterOpen) {

  shouldRoll = true;

  LOG.debug("BucketingSink {} starting new bucket.", subtaskIndex);

 } else {

  long writePosition = bucketState.writer.getPos();

  //根据文件偏移量来判断是否达到setBatchSize方法设定的滚动阈值

  if (writePosition > batchSize) {

   shouldRoll = true;

   LOG.debug(

    "BucketingSink {} starting new bucket because file position {} is above batch size {}.",

    subtaskIndex,

    writePosition,

    batchSize);

  }

  //根据时间来判断是否达到setInactiveBucketThreshold方法设定的滚动阈值

  else {

   if (currentProcessingTime - bucketState.creationTime > batchRolloverInterval) {

    shouldRoll = true;

    LOG.debug(

     "BucketingSink {} starting new bucket because file is older than roll over interval {}.",

     subtaskIndex,

     batchRolloverInterval);

   }

  }

 }

 return shouldRoll;

}

调用shouldRoll方法判断如果需要滚动文件,则调用openNewPartFile方法创建新文件,此方法主要分为以下步骤:

  • 调用closeCurrentPartFile方法关闭当前文件,核心操作就是将progress状态文件改为pedding状态文件
  • 调用assemblePartPath方法生成新文件名,此方法涉及到子任务索引、以及当前桶计数器概念,自行看源码
  • 创建progress状态文件,并打开流

讲完shouldRoll再讲下数据写入,invoke方法中数据写入只有简简单单一行:
bucketState.writer.write(value),我们先看一下bucketState对象中writer对象哪里来,整体还是比较绕的,分下面几步:

  • 业务代码中通过BucketingSink#setWriter方法设置writerTemplate属性
  • 在openNewPartFile方法中通过writerTemplate.duplicate创建实例

有了writer对象后,我们看一下实际写入代码,以平时最常用的StringWriter为例:

public void write(T element) throws IOException {

 //这里是直接调用HDFS文件流写入数据

 FSDataOutputStream outputStream = getStream();

 outputStream.write(element.toString().getBytes(charset));

 outputStream.write('\n');

}

0x5 文件状态流转

上一节只是完成了数据写入的分析,写入到 progress的文件是不能被HIVE加载查询的,Flink采用类型二阶段提交的来保证数据的一致性,状态流转是这样的:progress->pedding->finished
本节我们来分析一下是如来来完成文件状态流转的。
上一节在openNewPartFile方法源码分析中提到closeCurrentPartFile方法会把progress状态文件转为pedding状态文件,我们再来看一下源码:

private void closeCurrentPartFile(BucketState<T> bucketState) throws Exception {

 if (bucketState.isWriterOpen) {

  bucketState.writer.close();

  bucketState.isWriterOpen = false;

 }

 

 if (bucketState.currentFile != null) {

  Path currentPartPath = new Path(bucketState.currentFile);

  Path inProgressPath = getInProgressPathFor(currentPartPath);

  Path pendingPath = getPendingPathFor(currentPartPath);

 

  //重命名文件

  fs.rename(inProgressPath, pendingPath);

  //将文件加入到pedding列表中,snapshotState方法会用到

  bucketState.pendingFiles.add(currentPartPath.toString());

  bucketState.currentFile = null;

 }

}

从pedding状态到finished状态是又是如何做的呢?大家知道Flink是通过checkpoint机制来保证数据一致性,BucketingSink也是一样用了checkpoint来保证文件状态流转,确保最终数据一致性。
文章一开始类图处就已经提到重点关注的接口,其中一个是CheckpointedFunction,他有两个方法:

  • snapshotState:检查点触发时调用
  • initializeState:初始化时调用
    按一般正常思路,大家会觉得应该在snapshotState方法将pedding状态改为finished状态,不过BucketingSink做个小技巧,方法源码就不全贴了,核心代码如下:

bucketState.pendingFilesPerCheckpoint.put(context.getCheckpointId(), bucketState.pendingFiles);

这么做的目的只是让snapshotState方法快速完成,不影响其他流,实际状态流转放到了notifyCheckpointComplete方法中,此方法来自于CheckpointListener接口,当检查点完成时调用此方法,此方法具体源码不做分析,比较简单,将pedding后缀去掉完成重命名,这样一个文件的整体生命周期就结束了。

 

这篇关于Flink BucketingSink 源码分析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1024682

相关文章

Java程序进程起来了但是不打印日志的原因分析

《Java程序进程起来了但是不打印日志的原因分析》:本文主要介绍Java程序进程起来了但是不打印日志的原因分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Java程序进程起来了但是不打印日志的原因1、日志配置问题2、日志文件权限问题3、日志文件路径问题4、程序

Java 正则表达式URL 匹配与源码全解析

《Java正则表达式URL匹配与源码全解析》在Web应用开发中,我们经常需要对URL进行格式验证,今天我们结合Java的Pattern和Matcher类,深入理解正则表达式在实际应用中... 目录1.正则表达式分解:2. 添加域名匹配 (2)3. 添加路径和查询参数匹配 (3) 4. 最终优化版本5.设计思

Java字符串操作技巧之语法、示例与应用场景分析

《Java字符串操作技巧之语法、示例与应用场景分析》在Java算法题和日常开发中,字符串处理是必备的核心技能,本文全面梳理Java中字符串的常用操作语法,结合代码示例、应用场景和避坑指南,可快速掌握字... 目录引言1. 基础操作1.1 创建字符串1.2 获取长度1.3 访问字符2. 字符串处理2.1 子字

Python 迭代器和生成器概念及场景分析

《Python迭代器和生成器概念及场景分析》yield是Python中实现惰性计算和协程的核心工具,结合send()、throw()、close()等方法,能够构建高效、灵活的数据流和控制流模型,这... 目录迭代器的介绍自定义迭代器省略的迭代器生产器的介绍yield的普通用法yield的高级用法yidle

C++ Sort函数使用场景分析

《C++Sort函数使用场景分析》sort函数是algorithm库下的一个函数,sort函数是不稳定的,即大小相同的元素在排序后相对顺序可能发生改变,如果某些场景需要保持相同元素间的相对顺序,可使... 目录C++ Sort函数详解一、sort函数调用的两种方式二、sort函数使用场景三、sort函数排序

Java调用C++动态库超详细步骤讲解(附源码)

《Java调用C++动态库超详细步骤讲解(附源码)》C语言因其高效和接近硬件的特性,时常会被用在性能要求较高或者需要直接操作硬件的场合,:本文主要介绍Java调用C++动态库的相关资料,文中通过代... 目录一、直接调用C++库第一步:动态库生成(vs2017+qt5.12.10)第二步:Java调用C++

kotlin中const 和val的区别及使用场景分析

《kotlin中const和val的区别及使用场景分析》在Kotlin中,const和val都是用来声明常量的,但它们的使用场景和功能有所不同,下面给大家介绍kotlin中const和val的区别,... 目录kotlin中const 和val的区别1. val:2. const:二 代码示例1 Java

Go标准库常见错误分析和解决办法

《Go标准库常见错误分析和解决办法》Go语言的标准库为开发者提供了丰富且高效的工具,涵盖了从网络编程到文件操作等各个方面,然而,标准库虽好,使用不当却可能适得其反,正所谓工欲善其事,必先利其器,本文将... 目录1. 使用了错误的time.Duration2. time.After导致的内存泄漏3. jsO

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很

Spring事务中@Transactional注解不生效的原因分析与解决

《Spring事务中@Transactional注解不生效的原因分析与解决》在Spring框架中,@Transactional注解是管理数据库事务的核心方式,本文将深入分析事务自调用的底层原理,解释为... 目录1. 引言2. 事务自调用问题重现2.1 示例代码2.2 问题现象3. 为什么事务自调用会失效3