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