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#