播面

在利用 Flink CDC 实时同步高吞吐 MySQL 数据入 Paimon 的场景中，设置 15 秒的 Checkpoint 会带来极高的实时性，但这也是 HDFS 小文件暴涨的直接诱因。

在 Paimon 的 LSM 架构中，每次 Checkpoint 都会强制将内存中的数据 Flush 到 HDFS 产生新的物理文件（L0层）。在 15 秒的高频提交下，如果分桶（Bucket）较多或分区较散，极易导致 HDFS NameNode 元数据压力暴增，且伴随 Compaction 抢占 CPU/IO 带来的 Flink 任务背压。

在不改变 15 秒 Checkpoint 间隔的前提下，可以通过优化 Paimon 的写入、Compaction、缓冲及分桶参数，来有效平衡小文件与写入性能。

一、 解耦 Compaction 与写入（避免写背压与卡顿）

默认情况下，Paimon 的 Writer 算子在写入的同时会在后台线程执行 Compaction。但在高吞吐、高时效场景下，Compaction 消耗的 CPU/IO 会直接导致 Checkpoint 超时或引发写入背压。

根据所选的分桶模式（Bucket Mode），有以下两种解耦策略：

方案 A：开启 write-only + 提交独立的 Compaction 作业（推荐用于 Fixed Bucket）

如果使用的是固定分桶（bucket > 0），推荐将 Compaction 彻底从主写入任务中剥离：

1. 主写入任务配置：在建表语句的 WITH 参数中设置 'write-only' = 'true'。此时，Flink 写入任务只负责写数据并生成 Snapshot，完全不进行任何 Compaction 和 Snapshot 过期清理，从而保证 15 秒的 Checkpoint 极速完成且绝不被阻塞。
2. 启动独立 Compaction 任务：通过 Flink Action 或 Flink SQL 提交一个专用的后台 Compaction 任务，对该表进行持续的异步合并：
   sql

   -- 提交独立的流式 Compaction 任务
   CALL sys.compact(
     `table` => 'default.your_table_name',
     options => 'sink.parallelism=4'
   );

   *注意：若使用 Dynamic Bucket（bucket = -1），Paimon 暂不支持多任务写入同一分区，因此该场景下不能使用 write-only 与独立 Compaction 作业，须采用方案 B。*

方案 B：配置完全异步、非阻塞的 inline Compaction（推荐用于 Dynamic Bucket）

若无法使用独立作业，必须在写入任务中进行 inline Compaction，应调整参数使其“完全异步”，避免因 sorted run 堆积而触发“暂停写入（Write Stall）”：

• num-sorted-run.stop-trigger：设为一个极大值（如 2147483647 或 30，默认是 compaction-trigger + 3）。当未合并的小文件增多时，该配置能防止 Writer 主动挂起或暂停写入。
• sort-spill-threshold：设为 10，控制排序溢写的阈值。
• changelog-producer.lookup-wait（若启用了 lookup 类型的 Changelog 产生器）：设为 false，查找旧值时不阻塞写入。

二、 开启 Changelog 小文件合并（针对 CDC 场景）

Flink CDC 同步经常需要向下游流式发送 Changelog。如果配置了 'changelog-producer' = 'input' / 'lookup' / 'full-compaction'，在 15 秒 Checkpoint 下，除了产生数据小文件，还会产生海量的 Changelog 小文件。

• 参数配置：'precommit-compact' = 'true'（或在较新版本中为 'changelog.precommit-compact'）。
• 优化原理：在 Flink Writer 算子写入之后、正式 Commit 之前，Paimon 会在拓扑中自动加入 Compact Coordinator 与 Compact Worker。它们会在提交前把内存或临时生成的小 Changelog 文件预先合并（Pre-commit Merge）成大文件，从源头上斩断 HDFS 上 Changelog 小文件的生成。

三、 写入缓冲区（Buffer）调优（避免提前 Spill 产生额外小文件）

在 15 秒的 Checkpoint 间隔内，如果 Flink 算子内存中的 Buffer 提前满了，即使还没到 Checkpoint 时间，Paimon 也会被迫将 Buffer 中的数据 Spill（溢写）到 HDFS，从而额外产生碎小文件。

• 优化策略：
  1. 增大 Buffer：设置 'write-buffer-size' = '256 mb'（或更大，根据每 15 秒的实际吞吐量调整），确保 Buffer 足以容纳 15 秒内产生的所有数据，尽量只在 Checkpoint 触发时做单次 Flush。
  2. 启用 Flink 托管内存：设置 'sink.use-managed-memory-allocator' = 'true'。这将使用 Flink TaskManager 的 Managed Memory（管理内存）来存放 Paimon 的写入 Buffer，相比 JVM 堆内存，它在多任务共享和稳定性上表现更好，能有效降低 OOM 风险并平滑 Spill 写入。
  3. 保证 write-buffer-spillable = true（默认开启）：作为安全垫，防止瞬时流量激增导致 OOM。

四、 分桶（Bucket）与并行度调优（控制批次文件物理膨胀）

小文件的物理个数计算公式通常为：当日活跃分区数 × 单分区 Bucket 数 × Sink 并行度。在高吞吐下，Bucket 划分不合理是小文件暴涨的元凶。

1. 合理缩减 Bucket 数量：
   • Paimon 官方建议每个 Bucket 的健康数据量在 200MB ~ 1GB 之间。
   • 很多用户在建表时习惯性设置几十甚至上百个 Bucket，如果 15s 内只有少量数据流入，这上百个 Bucket 就会在 Checkpoint 时各自生成一个小文件。请结合日增量和时效性，将 Bucket 压缩至合理的个位数或采用 bucket = -1 动态分桶。
2. 对齐 Sink 并行度：
   • 设置 sink.parallelism 等于该表的 Bucket 数量（如果是固定分桶模式）。避免设置过大的并发，导致数据被过度稀释到不同的 Task 从而产生零碎小文件。

五、 其他辅助优化

1. 选择高压缩比、低开销的压缩格式：
   • 'file.compression' = 'zstd'
   • 'file.compression.zstd-level' = '3'
   • Zstd 能在较低 CPU 开销下提供极高的压缩率，有助于减少每次 Flush 和 Compaction 时的 HDFS I/O 字节量。
2. 将无特殊 OLAP 点查需求的表格式设为 Avro：
   • 如果该表仅作为 ODS 贴源层，后续不需要高频的列式 Ad-hoc 查询，可以考虑将 'file.format' = 'avro'。Avro 作为行存格式，其写入和序列化开销显著低于 Parquet/ORC，能极大释放 Flink 写入时的 CPU 压力。
3. 调整小文件判别比例：
   • 通过 'compaction.small-file-ratio' = '0.3'（默认 0.5）来微调，避免 Paimon 过于频繁地去 Compact 那些“虽然不满足 target-file-size、但已经接近该大小”的文件，从而减少写放大。

总结配置示例（DDL WITH 参数参考）

CREATE TABLE mysql_sync_paimon (
    id BIGINT,
    data STRING,
    dt STRING,
    PRIMARY KEY (dt, id) NOT ENFORCED
) PARTITIONED BY (dt)
WITH (
    'bucket' = '4',                            -- 适当缩减 bucket 数量，防止小文件散落
    'sink.parallelism' = '4',                  -- 与 bucket 保持一致
    
    -- 写入缓冲区与内存优化
    'write-buffer-size' = '256 mb',            -- 增大 buffer 避免非 checkpoint 期间溢写
    'sink.use-managed-memory-allocator' = 'true', -- 使用 Flink 托管内存
    
    -- 异步与非阻塞 Compaction 优化
    'num-sorted-run.stop-trigger' = '2147483647', -- 彻底防止 Compaction 慢导致写暂停
    'sort-spill-threshold' = '10',
    
    -- Changelog 预合并（CDC 降小文件利器）
    'changelog-producer' = 'input',            -- 视下游流读情况而定
    'precommit-compact' = 'true',              -- 开启 commit 前的 changelog 小文件合并
    
    -- 文件和压缩
    'file.format' = 'parquet',                 -- 若无列存需求，甚至可换成 'avro' 以降低写入开销
    'file.compression' = 'zstd',               -- 兼顾高压缩与低 CPU 开销
    'file.compression.zstd-level' = '3'
);