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，官方已经修复见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#