源码分析：Flink FileSystemTableSink 中自定义分区策略

源码分析 - Flink FileSystemTableSink 中自定义分区策略的创建与调用机制详解

在使用 Flink 通过 Table API 或 SQL 写入文件系统时，若启用了自定义分区，系统如何注入并应用参数 sink.partition-commit.policy.kind 来控制分区提交行为？本文将从源码角度梳理这套机制的调用路径与实现细节。

本人实现的 [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy https://github.com/apache/flink/pull/22831 也是基于本篇文章的源码分析而实现的。

示例：

# 配置双策略链（元数据提交+自定义）
sink.partition-commit.policy.kind: custom

# 自定义策略配置
sink.partition-commit.policy.custom.class: com.company.CustomPolicy
sink.partition-commit.policy.custom.parameters: param1,param2

---

一、调用链概览：参数作用于哪里？

当用户提交 Flink SQL 作业，FileSystemTableSink 会作为 TableSink 参与物理执行计划的构建。其核心的参数处理逻辑发生在如下调用链中：

translateToPlanInternal
 └── createSinkTransformation
     └── getSinkRuntimeProvider
         └── consumeDataStream
             └── consume
                 └── createStreamingSink
                     └── sink（创建 PartitionCommitter）
                         └── initializeState（构建策略链）

---

二、物理计划转换：从 Plan 到 Sink 的转化

Flink Planner 在构造物理执行计划时（Transformation 或 DataStream），会调用 TableSink 的 translateToPlanInternal 方法。在流式写入文件系统时，最终会进入 consume 方法中：

@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context sinkContext) {
    return new DataStreamSinkProvider() {
        @Override
        public DataStreamSink<?> consumeDataStream(
                ProviderContext providerContext, DataStream<RowData> dataStream) {
            return consume(providerContext, dataStream, sinkContext);
        }
    };
}

通过“数据是否有界”来判断是否是流处理（unbounded）：

• 若为流模式，则进入 createStreamingSink，构建 FileSystemSink；
• 否则使用 createBatchSink 执行批量输出。

---

三、策略链构建核心：createStreamingSink -> sink

createStreamingSink 中最关键的调用是 sink() 方法，负责将数据写入文件系统并执行分区提交策略，保证数据一致性和可见性。

它负责最终创建分区提交组件 PartitionCommitter，并将其与 writerStream 进行连接：

public static DataStreamSink<?> sink(
    ProviderContext providerContext,
    DataStream<PartitionCommitInfo> writerStream,
    Path path,
    ObjectIdentifier tableIdentifier,
    List<String> partitionKeys,
    TableMetaStoreFactory metaStoreFactory,
    FileSystemFactory fsFactory,
    ReadableConfig tableOptions
) {
    // 参数判断：是否启用分区提交策略
    if (partitionKeys.size() > 0 && options.contains(SINK_PARTITION_COMMIT_POLICY_KIND)) {
        // 构建提交器
        PartitionCommitter committer = new PartitionCommitter(
            path, tableIdentifier, partitionKeys, metaStoreFactory, fsFactory, tableOptions.toConfiguration()
        );
        return writerStream.transform("Partition Committer", committer);
    }
}

---

四、运行时策略链构建机制：PartitionCommitter

在 Flink Job 启动或恢复时，PartitionCommitter.initializeState() 被自动调用：

public void initializeState(StateInitializationContext context) throws Exception {
    PartitionCommitPolicyFactory factory = new PartitionCommitPolicyFactory(
        conf.get(SINK_PARTITION_COMMIT_POLICY_KIND),     // 如 "metastore,success-file"
        conf.get(SINK_PARTITION_COMMIT_POLICY_CLASS),     // 自定义类名（可选）
        conf.get(SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME), // success-file 策略用
        conf.get(SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS) // 自定义类参数（可选）
    );
    this.policies = factory.createPolicyChain(
        getUserCodeClassloader(), // 用户类加载器
        () -> { // 获取文件系统实例的工厂
            try {
                return fsFactory.create(locationPath.toUri());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    );
}

---

五、策略工厂 PartitionCommitPolicyFactory 的核心逻辑

策略链的创建会用到用户自定义类的类加载器（UserCodeClassloader）和文件系统工厂（FileSystem factory），确保策略能够正确加载用户代码并操作分区文件系统。

/** Create a policy chain. */
public List<PartitionCommitPolicy> createPolicyChain(
        ClassLoader cl, Supplier<FileSystem> fsSupplier) {
    String[] policyStrings = policyKind.split(",");
    return Arrays.stream(policyStrings)
            .map(
                    name -> {
                        switch (name.toLowerCase()) {
                            case PartitionCommitPolicy.METASTORE:
                                return new MetastoreCommitPolicy();
                            case PartitionCommitPolicy.SUCCESS_FILE:
                                return new SuccessFileCommitPolicy(
                                        successFileName, fsSupplier.get());
                            case PartitionCommitPolicy.CUSTOM:
                                try {
                                    if (parameters != null && !parameters.isEmpty()) {
                                        String[] paramStrings =
                                                parameters.toArray(new String[0]);
                                        Class<?>[] classes = new Class<?>[parameters.size()];
                                        for (int i = 0; i < parameters.size(); i++) {
                                            classes[i] = String.class;
                                        }
                                        return (PartitionCommitPolicy)
                                                cl.loadClass(customClass)
                                                        .getConstructor(classes)
                                                        .newInstance((Object[]) paramStrings);
                                    } else {
                                        return (PartitionCommitPolicy)
                                                cl.loadClass(customClass).newInstance();
                                    }}}}})
            .collect(Collectors.toList());}
}

关键点：自定义类必须实现 PartitionCommitPolicy 接口，并提供合适的构造方法（带参或无参均可）。

---

总结

参数如 sink.partition-commit.policy.kind = metastore,success-file 的注入生效流程如下：

用户配置
  ↓
FileSystemTableSink -> sink()
  ↓
构造 PartitionCommitter
  ↓
initializeState()
  ↓
PartitionCommitPolicyFactory -> createPolicyChain()
  ↓
创建策略实例 → 构成策略链 → 实现 PartitionCommitter