背景
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，官方已经修复见FLINK-20213。理论上每次Checkpoint都应该提交。
1.12
由于用户的呼声很高，官方也考虑到这个特性并在1.12中实现了，见FLINK-19345。
实现原理
增加了两个参数配置：
- auto-compaction 是否自动合并
- compaction.file-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个文件直接重命名，超过一个的进行合并。
再后面就是和1.11一样的处理逻辑了。
详细设计文档参考：https://docs.google.com/document/d/1cdlyoqgBq9yJEiHFBziimIoKHapQiEY2-0Tn8IF6G-c/edit#
