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
后缀去掉完成重命名,这样一个文件的整体生命周期就结束了。
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
即学即用的 30 段 Python 实用代码
Python是目前最流行的语言之一,它在数据科学、机器学习、web开发、脚本编写、自动化方面被许多人广泛使用。它的简单和易用性造就了它如此流行的原因。在本文中,我们将会介绍 30 个简短的代码片段,你可以在 30 秒或更短的时间里理解和学习这些代码片段。 1.检查重复元素 下面的方法可以检查给定列表中是否有重复的元素。它使用了 set() 属性,该属性将会从列表中删除重复的元素。 def all_unique(lst): return len(lst) == len(set(lst)) x = [1,1,2,2,3,2,3,4,5,6] y = [1,2,3,4,5] all_unique(x) # False all_unique(y) # True 2.变位词 检测两个字符串是否互为变位词(即互相颠倒字符顺序) ''' 遇到问题没人解答?小编创建了一个Python学习交流QQ群:857662006 寻找有志同道合的小伙伴,互帮互助, 群里还有不错的视频学习教程和PDF电子书! ''' from collections import Counter def anagram(first...
- 下一篇
【从入门到放弃-ZooKeeper】ZooKeeper实战-分布式锁
前言 上文【从入门到放弃-ZooKeeper】ZooKeeper实战-分布式队列中,我们一起写了下如何通过ZooKeeper的持久性顺序节点实现一个分布式队列。本文我们来一起写一个ZooKeeper的实现的分布式锁。 设计 参考之前学习的【从入门到放弃-Java】并发编程-JUC-locks-ReentrantLock,实现java.util.concurrent.locks.Lock接口。我们通过重写接口中的方法实现一个可重入锁。 lock:请求锁,如果成功则直接返回,不成功则阻塞 直到获取锁。 lockInterruptibly:请求锁,如果失败则一直阻塞等待 直到获取锁或线程中断 tryLock:1、尝试获取锁,获取失败的话 直接返回false,不会再等待。2、尝试获取锁,获取成功返回true,否则一直请求,直到超时返回false unl
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
-
Docker使用Oracle官方镜像安装(12C,18C,19C)
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8编译安装MySQL8.0.19
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- MySQL8.0.19开启GTID主从同步CentOS8
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
推荐阅读
最新文章
- Windows10,CentOS7,CentOS8安装Nodejs环境
- CentOS8编译安装MySQL8.0.19
- CentOS6,CentOS7官方镜像安装Oracle11G
- SpringBoot2整合Thymeleaf,官方推荐html解决方案
- 设置Eclipse缩进为4个空格,增强代码规范
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Windows10,CentOS7,CentOS8安装MongoDB4.0.16
- CentOS7安装Docker,走上虚拟化容器引擎之路
- CentOS6,7,8上安装Nginx,支持https2.0的开启
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题