1. 前言
Paimon的前身是Flink-Table-Store，希望提供流批一体的存储，提供一定的OLAP查询能力（基于列式存储），做到毫秒级别的实时流式读取。Flink-Table-Store希望能够支持Flink SQL的全部概念，能够结合Flink SQL提供DB级别体验，并且支持大规模的更新。Flink-Table-Store希望能够结合Flink，实现完整的流批一体体验（计算+存储），同时拓展Flink-Table-Store的生态，升级为Paimon，来支持更多大数据引擎的查询/写入。如果我们希望深度使用Paimon，并充分利用Paimon的特性，那么了解Flilnk写入Paimon的过程十分重要，本文希望通过源码分析的方式带大家充分了解Flink写入Paimon的完整过程。
2. 源码阅读demo
阅读源码最有效的方式就是跟读最简单的代码，由于Paimon的定位是打造类似Database的使用体验，与SQL结合很紧密，通常可以使用Flink SQL进行写入，但这里为了方便跟读代码，我们采用调api的方式进行写入
// 构建表
TableEnvironmentImpl tableEnv = TableEnvironmentImpl.create(EnvironmentSettings.newInstance().build());
String tableSql = String.format(
"create table %s (uuid string, name string, age int, ts int, part string, primary key (part,uuid) NOT ENFORCED) partitioned by (part) with ('connector' = 'paimon', 'bucket' = '2', 'bucket-key' = 'uuid', 'path' = '%s')"
, tableName
, filePath
);
tableEnv.executeSql(tableSql);
Catalog catalog = tableEnv.getCatalog(tableEnv.getCurrentCatalog()).get();
String database = catalog.getDefaultDatabase();
ResolvedCatalogTable catalogTable = (ResolvedCatalogTable) catalog.getTable(new ObjectPath(database, tableName));
ObjectIdentifier tablePath = ObjectIdentifier.of(catalogName, database, tableName);
FactoryUtil.DefaultDynamicTableContext context = new FactoryUtil.DefaultDynamicTableContext(
tablePath,
catalogTable,
Collections.emptyMap(),
Configuration.fromMap(catalogTable.getOptions()),
Thread.currentThread().getContextClassLoader(),
false
);
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(20000);
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
// 数据源
DataStream<RowData> dataStream = ...
// 将数据源写入Paimon表
((DataStreamSinkProvider) new FlinkTableFactory()
.createDynamicTableSink(context)
.getSinkRuntimeProvider(new SinkRuntimeProviderContext(false)))
.consumeDataStream(n -> Optional.empty(), dataStream);
env.execute("PaimonWriteDemoWithFlink");
3. Flink写入Paimon的完整流程
Paimon表的写入是通过FlinkTableSink实现DynamicTableSink接口来写入数据，核心逻辑位于getSinkRuntimeProvider方法中，后面我们会重点跟读getSinkRuntimeProvider方法中的内容。
这里也顺带讲一下通过Flink SQL进行写入的场景，对于Flink SQL而言，首先会通过calcite进行解析优化后生成JobGraph，再提交集群运行。我们希望了解Flink写入Paimon的完整过程，可以通过sql转化成JobGraph的过程中得到的Transformation列表去了解中间具体进行了哪些操作。下图是一条简单Flink SQL转化后的Transformation序列，其中红框内的Transformation序列是Paimon数据写入的完整过程，本质上是通过执行getSinkRuntimeProvider方法生成的。
接着，我们对Flink写入Paimon表的流程进行了梳理，整理出完整的类图，如下图所示
其中，核心入口FlinkTableSink的getSinkRuntimeProvider方法逻辑如下所示，
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
...
LogSinkProvider logSinkProvider = null;
if (logStoreTableFactory != null) {
// 当配置了log.system，则会生成一个log sink的提供器，目前只有kafka的日志存储
logSinkProvider = logStoreTableFactory.createSinkProvider(this.context, context);
}
Options conf = Options.fromMap(table.options());
// 生成log sink函数，当overwrite模式时，不配置log sink
final LogSinkFunction logSinkFunction =
overwrite ? null : (logSinkProvider == null ? null : logSinkProvider.createSink());
return new PaimonDataStreamSinkProvider(
(dataStream) ->
// FileStoreTable时FileStore的抽象层，提供InternalRow的读取和写入
new FlinkSinkBuilder((FileStoreTable) table)
.withInput(
new DataStream<>(dataStream.getExecutionEnvironment(),dataStream.getTransformation()))
.withLogSinkFunction(logSinkFunction)
.withOverwritePartition(overwrite ? staticPartitions : null)
.withParallelism(conf.get(FlinkConnectorOptions.SINK_PARALLELISM))
.build());
}
继续跟到FlinkSinkBuilder的build方法，会根据Paimon表的bucket模式生成不同类型的Sink
public DataStreamSink<?> build() {
BucketMode bucketMode = table.bucketMode();
switch (bucketMode) {
case FIXED:
// 生成固定bucket个数的Sink，这种方式结构固定，本文着重分析这一路径
return buildForFixedBucket();
case DYNAMIC:
// 生成动态变动Bucket个数的Sink，
return buildDynamicBucketSink();
case UNAWARE:
return buildUnawareBucketSink();
default:
throw new UnsupportedOperationException("Unsupported bucket mode: " + bucketMode);
}
}
private DataStreamSink<?> buildForFixedBucket() {
// 首先根据分区信息、bucket字段进行bucket分组
DataStream<RowData> partitioned =
partition(
input,
new RowDataChannelComputer(table.schema(), logSinkFunction != null),
parallelism);
// FileStoreSink实现将记录写入Paimon，FileStoreSink提供了生成写入算子的方法
FileStoreSink sink = new FileStoreSink(table, overwritePartition, logSinkFunction);
return sink.sinkFrom(partitioned);
}
接着继续跟读到FileStoreSink的sinkFrom的方法
public DataStreamSink<?> sinkFrom(DataStream<T> input) {
String initialCommitUser = UUID.randomUUID().toString();
return sinkFrom(input, initialCommitUser);
}
public DataStreamSink<?> sinkFrom(DataStream<T> input, String initialCommitUser) {
// 执行真正的写入操作，在这个阶段不会进行提交，相当于两阶段提交的第一阶段，进行数据写入，不会有snapshot生成
SingleOutputStreamOperator<Committable> written =
doWrite(input, initialCommitUser, input.getParallelism());
// 执行提交操作，会生成snapshot，下游可见，如果日志配置
return doCommit(written, initialCommitUser);
}
public SingleOutputStreamOperator<Committable> doWrite(
DataStream<T> input, String commitUser, Integer parallelism) {
...
// 是否只是writeOnly，如果是，则会忽略compact和snapshot过期，这个配置需要结合专门的compact任务执行，不然会造成小文件剧增，同时降低数据读的性能
Boolean writeOnly = table.coreOptions().writeOnly();
SingleOutputStreamOperator<Committable> written =
input.transform(
(writeOnly ? WRITER_WRITE_ONLY_NAME : WRITER_NAME) + " -> " + table.name(),
new CommittableTypeInfo(),
// 生成写入算子，上游的数据会通过这个算子将数据写入到Paimon表
createWriteOperator(
createWriteProvider(env.getCheckpointConfig(), isStreaming), commitUser)
).setParallelism(parallelism == null ? input.getParallelism() : parallelism);
...
Options options = Options.fromMap(table.options());
if (options.get(SINK_USE_MANAGED_MEMORY)) {
// Flink会创建一个独立的内存分配器用于merge tree的数据写入操作
// 否则会使用TM的管理内存支持写入操作
MemorySize memorySize = options.get(SINK_MANAGED_WRITER_BUFFER_MEMORY);
written.getTransformation()
.declareManagedMemoryUseCaseAtOperatorScope(
ManagedMemoryUseCase.OPERATOR, memorySize.getMebiBytes());
}
return written;
}
protected DataStreamSink<?> doCommit(DataStream<Committable> written, String commitUser) {
...
SingleOutputStreamOperator<?> committed =
written.transform(
GLOBAL_COMMITTER_NAME + " -> " + table.name(),
new CommittableTypeInfo(),
// 执行提交操作的算子，该算子会生成snapshot，下游可见
new CommitterOperator<>(
streamingCheckpointEnabled,
commitUser,
createCommitterFactory(streamingCheckpointEnabled),
createCommittableStateManager()))
.setParallelism(1)
.setMaxParallelism(1);
return committed.addSink(new DiscardingSink<>()).name("end").setParallelism(1);
}
至此，我们可以发现，Flink写入Paimon的所有Transformation已经构建完成了，接下来，我们可以跟着写入算子的processElement方法了解数据是如何成为Paimon底层的LSM-tree数据结构的。
4. Flink写入数据
Flink写入Paimon的算子是RowDataStoreWriteOperator，算子是预提交算子，会将数据flush的磁盘，但不会执行commit操作，核心代码如下
public void processElement(StreamRecord<RowData> element) throws Exception {
sinkContext.timestamp = element.hasTimestamp() ? element.getTimestamp() : null;
SinkRecord record;
try {
// 将数据写入Paimon的文件系统
record = write.write(new FlinkRowWrapper(element.getValue()));
} catch (Exception e) {
throw new IOException(e);
}
if (logSinkFunction != null) {
// 将数据双写到日志存储层，目前只支持写入到kafka
SinkRecord logRecord = write.toLogRecord(record);
logSinkFunction.invoke(logRecord, sinkContext);
}
}
上述算子中的write生成是通过FlieStoreTable.store().newWrite()的逻辑生成的
FileStoreTable是对Paimon表的文件存储的抽象层，提供底层数据的读写api，Paimon会根据WriteMode和是否包含主键来决定生成哪一种FileStoreTable：
- 当WriteMode = APPEND_ONLY，则会生成AppendOnlyFileStoreTable，这种表的写入流程很简单，当写入的记录数达到阈值且缓存数据量达到单文件大小时就会进行flush操作；
- 当WriteMode = CHANGE_LOG 且不包含主键，则会生成ChangelogValueCountFileStoreTable，当从内存池拉不到内存块的时候，会进行flush操作，在flush操作的时候，会记录flush的数据的统计值；
- 当WriteMode = CHANGE_LOG 且包含主键，则会生成ChangelogWithKeyFileStoreTable，与ChangelogValueCountFileStoreTable的flush时机一致，只是在flush中会根据主键进行merge操作；
- 当WriteMode = AUTO，则会根据是否有主键生成AppendOnlyFileStoreTable还是ChangelogWithKeyFileStoreTable。
接下来，我们会分析FileStoreTable为ChangelogWithKeyFileStoreTable的情况。根据图3-2的类图会发现，最终会调用到MergeTreeWriter的write()方法。该方法会先将数据缓存在内存缓存中（SortBuffer），该部分缓存数据为排序数据，当缓存满了之后，会将缓存中的数据flush到磁盘。
public void write(KeyValue kv) throws Exception {
long sequenceNumber =
kv.sequenceNumber() == KeyValue.UNKNOWN_SEQUENCE
? newSequenceNumber()
: kv.sequenceNumber();
// 将数据put到内存缓存中
boolean success = writeBuffer.put(sequenceNumber, kv.valueKind(), kv.key(), kv.value());
if (!success) {
// 缓存满了，会flush缓存数据到磁盘
flushWriteBuffer(false, false);
success = writeBuffer.put(sequenceNumber, kv.valueKind(), kv.key(), kv.value());
if (!success) {
throw new RuntimeException("Mem table is too small to hold a single element.");
}
}
}
flush的代码如下
private void flushWriteBuffer(boolean waitForLatestCompaction, boolean forcedFullCompaction) throws Exception {
if (writeBuffer.size() > 0) {
if (compactManager.shouldWaitForLatestCompaction()) {
// 判断是否需要进行等待上一次compact结束
waitForLatestCompaction = true;
}
// 根据changelogProducer的类型生成changelog文件写入器
final RollingFileWriter<KeyValue, DataFileMeta> changelogWriter =
changelogProducer == ChangelogProducer.INPUT ? writerFactory.createRollingChangelogFileWriter(0) : null;
// 生成flush数据的MergeTree文件写入器，根据文件大小进行rolling
final RollingFileWriter<KeyValue, DataFileMeta> dataWriter = writerFactory.createRollingMergeTreeFileWriter(0);
try {
// 将缓存中的所有数据flush到磁盘
writeBuffer.forEach(
keyComparator,
mergeFunction,
changelogWriter == null ? null : changelogWriter::write,
dataWriter::write);
} finally {
if (changelogWriter != null) {
changelogWriter.close();
}
dataWriter.close();
}
if (changelogWriter != null) {
// 将changelog新增文件缓存在算子中，供算子在进行checkpoint的时候将所有的flush下发到下游算子（提交算子），下发是在prepareSnapshotPreBarrier()方法中进行的，所以会在下游算子进行checkpoint之前接收所有的flush信息
newFilesChangelog.addAll(changelogWriter.result());
}
for (DataFileMeta fileMeta : dataWriter.result()) {
// 将新增的文件缓存在算子中，供算子在进行checkpoint的时候将所有的提交下发到下游算子（提交算子）
newFiles.add(fileMeta);
compactManager.addNewFile(fileMeta);
}
writeBuffer.clear();
}
// 尝试同步上一次compact结果
trySyncLatestCompaction(waitForLatestCompaction);
// 尝试去触发一次新的compact
compactManager.triggerCompaction(forcedFullCompaction);
}
在writeBuffer进行flush磁盘的时候，会对数据进行merge操作，当FileStoreTable为ChangelogWithKeyFileStoreTable时，会根据merge-engine配置来生成特定的merge引擎，也就是3-2类图中的merge function，具体代码如下，
...
// 获取merge-engine配置
CoreOptions.MergeEngine mergeEngine = options.mergeEngine();
KeyValueFieldsExtractor extractor = ChangelogWithKeyKeyValueFieldsExtractor.EXTRACTOR;
MergeFunctionFactory<KeyValue> mfFactory;
switch (mergeEngine) {
case DEDUPLICATE:
// merge操作时，只会保留最新的一条
mfFactory = DeduplicateMergeFunction.factory();
break;
case PARTIAL_UPDATE:
// 可以更新部分非空字段，可以设置字段级的sequence来进行局部比较更新
// 不支持retract类型的操作
mfFactory = PartialUpdateMergeFunction.factory(conf, rowType);
break;
case AGGREGATE:
// merge是对字段做聚合操作
mfFactory =
AggregateMergeFunction.factory(
conf,
tableSchema.fieldNames(),
rowType.getFieldTypes(),
tableSchema.primaryKeys());
break;
case FIRST_ROW:
// 只保留第一条
mfFactory =
FirstRowMergeFunction.factory(
new RowType(extractor.keyFields(tableSchema)), rowType);
break;
default:
throw new UnsupportedOperationException(
"Unsupported merge engine: " + mergeEngine);
}
if (options.changelogProducer() == ChangelogProducer.LOOKUP) {
// 当change log的生成方式为LOOKUP的时候，merge则为封装
mfFactory =
LookupMergeFunction.wrap(
mfFactory,
new RowType(extractor.keyFields(tableSchema)),
rowType);
}
在写入后，会进行尝试同步上一次compact的结果，同时也会尝试去触发一次新的compact，Paimon的Compact原理争取通过下一篇文章来分析。
这里，与Flink写入Hudi的过程一样，Flink写入Paimon是如何保证Exactly-Once语义的呢？
5. Flink提交写入状态
提交算子CommitterOperator首先会接收上游算子下发的flush信息
public void processElement(StreamRecord<CommitT> element) {
output.collect(element);
// 缓存上游下发的flush信息在算子内部
this.inputs.add(element.getValue());
}
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
// 在算子进行checkpoint完成后进行commit操作
commitUpToCheckpoint(endInput ? Long.MAX_VALUE : checkpointId);
}
private void commitUpToCheckpoint(long checkpointId) throws Exception {
NavigableMap<Long, GlobalCommitT> headMap =
committablesPerCheckpoint.headMap(checkpointId, true);
// 将算子内部缓存的flush信息整合进行commit操作
committer.commit(committables(headMap));
headMap.clear();
}
commiter.commit的逻辑是将所有的flush信息进行一次commit操作，最终flush信息会被序列化成json信息保存在snapshot文件中，当commit成功后，Paimon表会新增一次snapshot。详细的commit过程的代码比较简单，可以直接跟读，这里就不再赘述。
6. 最后
本文通过跟读源码的方式对Flink写入Paimon的核心流程进行了解析，相信通过对Flink写入Paimon流程细节的梳理，对理解Paimon的特性及性能优化都是有极大的助力。由于本人能力有限，文章中出现的错误在所难免，希望大家发现后帮忙指正，万分感谢。
最后总结一下，本文主要解析了Flink写入Paimon的核心流程：
1. 介绍了Flink SQL/api的方式构建写入流程DAG的完整过程；
2. 介绍了写入算子RowDataStoreWriteOperator处理数据的完整流程；
3. 介绍了提交算子CommitterOperator进行snapshot提交的完整流程。
当然，本文由于篇幅有限，没有对Flink和Paimon架构和概念进行详细的介绍，同时对Flink写入Paimon的Compact过程及性能优化也没有涉及，后续会加上这些方面的解析。
