2023年11月26日发(作者：)
Flink实战之合并⼩⽂件
背景
Flink的filesystem connector⽀持写⼊hdfs，同时⽀持基于Checkpoint的滚动策略，每次做Checkpoint时将inprogress的⽂件变为正
式⽂件，可供下游读取。
由于并⾏度设置、数据量⼤⼩、Checkpoint配置的不同、分区的选择，都有可能导致产⽣⼤量的⼩⽂件，这对hdfs产⽣很⼤影响。但是可
以通过⼀些⼿段来减少⼩⽂件，本⽂主要探讨⼀些filesystem connector⽀持的partition commit policy，通过⾃定义policy来合并⼩⽂
件。
1.11
⽬前1.11版本官⽅只⽀持metastore和success file的策略，以及可供⽤户⾃定义的custom策略。我们就⾃定义⼀个合并⼩⽂件的策略。
⽹上有现成的,我也是基于这个代码做了测试。
测试结果是写hdfs时可以正常合并⼩⽂件并⽣成⽬标⽂件，写hive表时却⽆法⽣成⽬标⽂件，具体原因还不得⽽知。
分区提交的原理
如果看过源码的话会发现filesystem是依赖StreamingFileSink来写⽂件的，更深⼊点就是StreamingFileWriter的Operator，如果指定了
分区提交策略还会有StreamingFileCommitter的Operator，分区提交的逻辑就在FileCommiter⾥⾯。
当Checkpoint完成最后会调⽤各个Operator的NotifyCheckpointComplete⽅法，在StreamingFileWriter的该⽅法中会构造⼀个
CommitMessage对象发送给下游Operator，CommitMessage包括CheckpointID、当前的subtask、所有subtask数量以及需要提交的
分区（⾮激活状态）。
private void commitUpToCheckpoint(long checkpointId) throws Exception {
helper.commitUpToCheckpoint(checkpointId);
CommitMessage message = new CommitMessage(
checkpointId,
getRuntimeContext().getIndexOfThisSubtask(),
getRuntimeContext().getNumberOfParallelSubtasks(),
new ArrayList<>(inactivePartitions));
output.collect(new StreamRecord<>(message));
inactivePartitions.clear();
}
由于StreamingFileCommitter的并发只有1，所以上游的StreamingFileWriter发送过来的CommitMessage都会交由同⼀个
StreamingFileCommitter处理，当判断收集到所有subtask的数据时，就认为可以进⾏分区提交了。然后分别执⾏各个policy的commit
⽅法。
private void commitPartitions(long checkpointId) throws Exception {
List<String> partitions = checkpointId == Long.MAX_VALUE ?
trigger.endInput() :
trigger.committablePartitions(checkpointId);
if (partitions.isEmpty()) {
return;
}
try (TableMetaStoreFactory.TableMetaStore metaStore = metaStoreFactory.createTableMetaStore()) {
for (String partition : partitions) {
LinkedHashMap<String, String> partSpec = extractPartitionSpecFromPath(new Path(partition));
LOG.info("Partition {} of table {} is ready to be committed", partSpec, tableIdentifier);
Path path = new Path(locationPath, generatePartitionPath(partSpec));
PartitionCommitPolicy.Context context = new PolicyContext(
new ArrayList<>(partSpec.values()), path);
for (PartitionCommitPolicy policy : policies) {
if (policy instanceof MetastoreCommitPolicy) {
((MetastoreCommitPolicy) policy).setMetastore(metaStore);
}
policy.commit(context);
}
}
}
}
这⾥⾯有个bug，官⽅已经修复见。理论上在分区创建和分区结束时分别提交⼀次（这⾥有个误解并不是每次Checkpoint都会提交，官⽹
说的⽴即提交的意思是分区⽣成时就提交⼀次，保证数据及时可查）。
1.12
由于⽤户的呼声很⾼，官⽅也考虑到这个特性并在1.12中实现了，见。
实现原理
增加了两个参数配置：
auto-compaction 是否⾃动合并
-size: compact target file size, default is rolling-file-size 合并后⽂件⼤⼩
增加了CompactFileWriter，并将之前的StreamingFileWriter变成了AbstractStreamingWriter抽象类。
在notifyCheckpointComplete⽅法中像下游发送EndCheckpoint对象
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
output.collect(new StreamRecord<>(new EndCheckpoint(
checkpointId,
getRuntimeContext().getIndexOfThisSubtask(),
getRuntimeContext().getNumberOfParallelSubtasks())));
}
增加了CompactCoordinator Operator，在这⾥对当前Checkpoint写⼊的分区的所有待合并⽂件进⾏分组，按照⽬标⽂件⼤⼩分组。
举个例⼦：
待合并⽂件有：
a 5M b 4M c 7M d 11M e 2M f 3M g 4M ,⽬标⽂件⼤⼩10M
合并后的结果是
a,b⼀组，c⼀组，d⼀组，e,f,g⼀组，并对分组进⾏编号0-ab,1-c,2-d,3-e,f,g
然后将分区的结果⼴播到下游CompactOperator。
在CompactOperator中对分组的⽂件进⾏合并，每个subtask只处理其中的部分，分配规则是编号%subtask总数。
如果分组中只有1个⽂件时通过bytes复制，超过⼀个的进⾏合并。
再后⾯就是和1.11⼀样的处理逻辑了。
详细设计⽂档参考：/document/d/1cdlyoqgBq9yJEiHFBziimIoKHapQiEY2-0Tn8IF6G-c/edit#
发布评论