Flink BucketingSink 源码分析

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 0x1 摘要 BucketingSink类提供了非常完美的功能支持数据落HDFS,在实际业务中不建议自己去实现,直接采用此类可以避免一些坑。注:此文基于Flink 1.6.3 版本源码。 0x2 BucketingSink 类结构分析 我们关注RichSinkFunction、Checkpoint.

0x1 摘要

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

0x2 BucketingSink 类结构分析

8306d3dd_78f7_437f_a496_62202542f0f8
我们关注RichSinkFunctionCheckpointedFunctionCheckpointListener三个父类

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后缀去掉完成重命名,这样一个文件的整体生命周期就结束了。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
5月前
|
Java 流计算
Flink中异步AsyncIO的实现 (源码分析)
Flink中异步AsyncIO的实现 (源码分析)
|
12月前
|
SQL API 流计算
Flink SQL代码补全提示(源码分析)
Flink SQL代码补全提示(源码分析)
68 0
|
SQL 存储 缓存
Flink进行Paimon写入源码分析
本文主要解析了Flink写入Paimon的核心流程。
|
存储 消息中间件 缓存
Flink进行Hudi写入源码分析
本文主要解析了Flink将DataStream写入到Hudi表的核心流程
|
SQL API Apache
Flink SQL代码补全提示(源码分析)
使用过Navicat的童鞋都知道,当我们写SQL的时候,工具会根据我们输入的内容弹出提示,这样可以很方便我们去写SQL
435 0
Flink SQL代码补全提示(源码分析)
|
SQL Java API
Flink 1.13.0 sql-client 新特性及源码分析
在 Flink 1.13.0 版本中增加了很多新特征,具体可以参考前面一篇文章,其中很重要的一点是对 sql-client 功能做了加强,支持了初始化脚本和执行 SQL 文件,SQL 客户端是直接运行和部署 SQL 流和批处理作业的便捷方法,而无需从命令行或作为 CI 的一部分来编写任何代码,这个版本大大改进了 SQL 客户端的功能。现在,SQL 客户端和SQL 脚本都支持 Java 应用程序可用的几乎所有操作(通过编程方式从 TableEnvironment 启动查询时)。这意味着 SQL 用户在其 SQL 部署中需要粘贴的代码变的更少.由于篇幅的原因这篇文章只会介绍 SQL CLIENT
Flink 1.13.0 sql-client 新特性及源码分析
|
Apache 调度 流计算
【Flink】(05)Apache Flink 漫谈系列 —— SocketWindowWordCount 程序执行过程源码分析3
【Flink】(05)Apache Flink 漫谈系列 —— SocketWindowWordCount 程序执行过程源码分析3
243 0
【Flink】(05)Apache Flink 漫谈系列 —— SocketWindowWordCount 程序执行过程源码分析3
|
分布式计算 数据处理 API
【Flink】(05)Apache Flink 漫谈系列 —— SocketWindowWordCount 程序执行过程源码分析1
【Flink】(05)Apache Flink 漫谈系列 —— SocketWindowWordCount 程序执行过程源码分析1
247 0
【Flink】(05)Apache Flink 漫谈系列 —— SocketWindowWordCount 程序执行过程源码分析1
|
SQL 缓存 JSON
Java SPI 机制在 Flink 中的应用(源码分析)
我们在使用 Flink SQL 的时候是否有过这样的疑问? Flink 提供了各种各样的 connector 我们只需要在 DML 里面定义即可运行,那它是怎么找到要执行的代码呢? 它是怎么知道代码对应关系的呢? 其实 Flink 是通过 Java 的 SPI(并不是Flink发明创造的) 机制来实现的,下面就来深入源码分析一下其实现过程. 什么是 SPI ?
Java SPI 机制在 Flink 中的应用(源码分析)
|
存储 流计算
Flink源码分析:WindowOperator底层实现
上一篇文章介绍了 Flink窗口机制的执行流程,其实WindowOperator才是真正负责window中元素存储和计算流程的核心类。这篇文章主要就是分析一下WindowOperator的执行逻辑。 apply方法 接着上一篇从apply方法入手,先来看一下apply的代码逻辑。