Apache Paimon核心原理和Flink应用进阶

这是一篇较为完整的介绍Apache Paimon和Flink进阶应用的文章，你最好收藏一波。

1.1 简介

Flink 社区希望能够将 Flink 的 Streaming 实时计算能力和 Lakehouse 新架构优势进一步结合，推出新一代的 Streaming Lakehouse 技术，促进数据在数据湖上真正实时流动起来，并为用户提供实时离线一体化的开发体验。Flink 社区内部孵化了 Flink Table Store （简称 FTS ）子项目，一个真正面向 Streaming 以及 Realtime的数据湖存储项目。2023年3月12日，FTS进入 Apache 软件基金会 (ASF) 的孵化器，改名为 Apache Paimon (incubating)。

Apache Paimon是一个流数据湖平台，具有高速数据摄取、变更日志跟踪和高效的实时分析的能力。

1. 读/写：Paimon 支持多种读/写数据和执行 OLAP 查询的方式。

（1）对于读取，它支持以下方式消费数据

从历史快照（批处理模式）、从最新的偏移量（在流模式下），或以混合方式读取增量快照。

（2）对于写入，它支持来自数据库变更日志（CDC）的流式同步或来自离线数据的批量插入/覆盖。

1. 生态系统

除了Apache Flink之外，Paimon还支持Apache Hive、Apache Spark、Trino等其他计算引擎的读取。

1. 内部

在底层，Paimon 将列式文件存储在文件系统/对象存储上，并使用 LSM 树结构来支持大量数据更新和高性能查询。

1. 统一存储

对于 Apache Flink 这样的流引擎，通常有三种类型的连接器：

1. 消息队列：例如 Apache Kafka，在源阶段和中间阶段都使用它，以保证延迟保持在秒级
2. OLAP系统：例如Clickhouse，它以流方式接收处理后的数据并为用户的即席查询提供服务
3. 批量存储：例如Apache Hive，它支持传统批处理的各种操作，包括INSERT OVERWRITE

Paimon 提供表抽象。它的使用方式与传统数据库没有什么区别：

1. 在批处理执行模式下，它就像一个Hive表，支持Batch SQL的各种操作。查询它以查看最新的快照。
2. 在流执行模式下，它的作用就像一个消息队列。查询它的行为就像从历史数据永不过期的消息队列中查询流更改日志。

1.2 核心特性

1）统一批处理和流处理

批量写入和读取、流式更新、变更日志生成，全部支持。

2）数据湖能力

低成本、高可靠性、可扩展的元数据。Apache Paimon 具有作为数据湖存储的所有优势。

3）各种合并引擎

按照您喜欢的方式更新记录。保留最后一条记录、进行部分更新或将记录聚合在一起，由您决定。

4）变更日志生成

Apache Paimon 可以从任何数据源生成正确且完整的变更日志，从而简化您的流分析。

5）丰富的表类型

除了主键表之外，Apache Paimon还支持append-only表，提供有序的流式读取来替代消息队列。

6）模式演化

Apache Paimon 支持完整的模式演化。您可以重命名列并重新排序。

1.3 基本概念

• 1.3.1 Snapshot

快照捕获表在某个时间点的状态。用户可以通过最新的快照来访问表的最新数据。通过时间旅行，用户还可以通过较早的快照访问表的先前状态。

• 1.3.2 Partition

Paimon 采用与 Apache Hive 相同的分区概念来分离数据。

分区是一种可选方法，可根据日期、城市和部门等特定列的值将表划分为相关部分。每个表可以有一个或多个分区键来标识特定分区。

通过分区，用户可以高效地操作表中的一片记录。

如果定义了主键，则分区键必须是主键的子集。

• 1.3.3 Bucket

未分区表或分区表中的分区被细分为存储桶，以便为可用于更有效查询的数据提供额外的结构。

桶的范围由记录中的一列或多列的哈希值确定。用户可以通过提供bucket-key选项来指定分桶列。如果未指定bucket-key选项，则主键（如果已定义）或完整记录将用作存储桶键。

桶是读写的最小存储单元，因此桶的数量限制了最大处理并行度。不过这个数字不应该太大，因为它会导致大量小文件和低读取性能。一般来说，建议每个桶的数据大小为1GB左右。

• 1.3.4 Consistency Guarantees一致性保证

Paimon writer使用两阶段提交协议以原子方式将一批记录提交到表中。每次提交在提交时最多生成两个快照。

对于任意两个同时修改表的writer，只要他们不修改同一个存储桶，他们的提交都是可序列化的。如果他们修改同一个存储桶，则仅保证快照隔离。也就是说，最终表状态可能是两次提交的混合，但不会丢失任何更改。

1.4 文件布局

一张表的所有文件都存储在一个基本目录下。Paimon 文件以分层方式组织。下图说明了文件布局。从快照文件开始，Paimon 读者可以递归地访问表中的所有记录。

下面简单介绍文件布局。

• 1.4.1 Snapshot Files

所有快照文件都存储在快照目录中。

快照文件是一个 JSON 文件，包含有关此快照的信息，包括：

正在使用的Schema文件

包含此快照的所有更改的清单列表（manifest list）

• 1.4.2 Manifest Files

所有清单列表（manifest list）和清单文件（manifest file）都存储在清单（manifest）目录中。

清单列表（manifest list）是清单文件名（manifest file）的列表。

清单文件（manifest file）是包含有关 LSM 数据文件和更改日志文件的文件信息。例如对应快照中创建了哪个LSM数据文件、删除了哪个文件。

• 1.4.3 Data Files

数据文件按分区和存储桶分组。每个存储桶目录都包含一个 LSM 树及其变更日志文件。目前，Paimon 支持使用 orc（默认）、parquet 和 avro 作为数据文件格式。

• 1.4.4 LSM Trees

Paimon 采用 LSM 树（日志结构合并树）作为文件存储的数据结构。

Sorted Runs

LSM 树将文件组织成多个Sorted Run。Sorted Run由一个或多个数据文件组成，并且每个数据文件恰好属于一个Sorted Run。

数据文件中的记录按其主键排序。在Sorted Run中，数据文件的主键范围永远不会重叠。

正如您所看到的，不同的Sorted Run可能具有重叠的主键范围，甚至可能包含相同的主键。查询LSM树时，必须合并所有Sorted Run，并且必须根据用户指定的合并引擎和每条记录的时间戳来合并具有相同主键的所有记录。

写入LSM树的新记录将首先缓存在内存中。当内存缓冲区满时，内存中的所有记录将被排序并刷新到磁盘。

Compaction

当越来越多的记录写入LSM树时，Sorted Run的数量将会增加。由于查询LSM树需要将所有Sorted Run合并起来，太多Sorted Run将导致查询性能较差，甚至内存不足。

为了限制Sorted Run的数量，我们必须偶尔将多个Sorted Run合并为一个大的Sorted Run。这个过程称为Compaction。

然而，Compaction是一个资源密集型过程，会消耗一定的CPU时间和磁盘IO，因此过于频繁的Compaction可能会导致写入速度变慢。这是查询和写入性能之间的权衡。Paimon 目前采用了类似于 Rocksdb 通用压缩的Compaction策略。

默认情况下，当Paimon将记录追加到LSM树时，它也会根据需要执行Compaction。用户还可以选择在“专用Compaction作业”中独立执行所有Compaction。

2.1 集成Flink进阶

2.1.1 写入性能

Paimon的写入性能与检查点密切相关，因此需要更大的写入吞吐量：

增加检查点间隔，或者仅使用批处理模式。
增加写入缓冲区大小。
启用写缓冲区溢出。
如果您使用固定存储桶模式，请重新调整存储桶数量。

• 并行度

建议sink的并行度小于等于bucket的数量，最好相等。

• Compaction

当Sorted Run数量较少时，Paimon writer 将在单独的线程中异步执行压缩，因此记录可以连续写入表中。然而，为了避免Sorted Runs的无限增长，当Sorted Run的数量达到阈值时，writer将不得不暂停写入。下表属性确定阈值。

当 num-sorted-run.stop-trigger 变大时，写入停顿将变得不那么频繁，从而提高写入性能。但是，如果该值变得太大，则查询表时将需要更多内存和 CPU 时间。如果您担心内存 OOM，请配置sort-spill-threshold。它的值取决于你的内存大小。

• 优先考虑写入吞吐量

如果希望某种模式具有最大写入吞吐量，则可以缓慢而不是匆忙地进行Compaction。可以对表使用以下策略

num-sorted-run.stop-trigger = 2147483647
sort-spill-threshold = 10

此配置将在写入高峰期生成更多文件，并在写入低谷期逐渐合并到最佳读取性能。

• 触发Compaction的Sorted Run数

Paimon使用LSM树，支持大量更新。LSM 在多次Sorted Runs中组织文件。从 LSM 树查询记录时，必须组合所有Sorted Runs以生成所有记录的完整视图。

过多的Sorted Run会导致查询性能不佳。为了将Sorted Run的数量保持在合理的范围内，Paimon writers 将自动执行Compaction。下表属性确定触发Compaction的最小Sorted Run数。

• 写入初始化

在write初始化时，bucket的writer需要读取所有历史文件。如果这里出现瓶颈（例如同时写入大量分区），可以使用write-manifest-cache缓存读取的manifest数据，以加速初始化。

• 内存

Paimon writer中主要占用内存的地方有3个：

Writer的内存缓冲区，由单个任务的所有Writer共享和抢占。该内存值可以通过 write-buffer-size 表属性进行调整。

合并多个Sorted Run以进行Compaction时会消耗内存。可以通过 num-sorted-run.compaction-trigger 选项进行调整，以更改要合并的Sorted Run的数量。

如果行非常大，在进行Compaction时一次读取太多行数据可能会消耗大量内存。减少 read.batch-size 选项可以减轻这种情况的影响。

写入列式（ORC、Parquet等）文件所消耗的内存，不可调。

2.2.2 读取性能

• Full Compaction

配置full-compaction.delta-commits在Flink写入中定期执行full-compaction。并且可以确保在写入结束之前分区被完全Compaction。

注意：Paimon 默认处理小文件并提供良好的读取性能。请不要在没有任何要求的情况下配置此Full Compaction选项，因为它会对性能产生重大影响。

• 主键表

对于主键表来说，这是一种MergeOnRead技术。读取数据时，会合并多层LSM数据，并行数会受到桶数的限制。虽然Paimon的merge会高效，但是还是赶不上普通的AppendOnly表。

如果你想在某些场景下查询得足够快，但只能找到较旧的数据，你可以：

配置full-compaction.delta-commits，写入数据时（目前只有Flink）会定期进行full Compaction。

配置`scan.mode`为`compacted-full`，读取数据时，选择full-compaction的快照。读取性能良好。

• 仅追加表

小文件会降低读取速度并影响 DFS 稳定性。默认情况下，当单个存储桶中的小文件超过“compaction.max.file-num”（默认50个）时，就会触发compaction。但是当有多个桶时，就会产生很多小文件。

您可以使用full-compaction来减少小文件。full-compaction将消除大多数小文件。

• 格式

Paimon 对 parquet 读取进行了一些查询优化，因此 parquet 会比 orc 稍快一些。

2.2.3 多Writer并发写入

Paimon的快照管理支持向多个writer写入。

默认情况下，Paimon支持对不同分区的并发写入。推荐的方式是streaming job将记录写入Paimon的最新分区；同时批处理作业（覆盖）将记录写入历史分区。

如果需要多个Writer写到同一个分区，事情就会变得有点复杂。例如，不想使用 UNION ALL，那就需要有多个流作业来写入"partial-update"表。参考如下的"Dedicated Compaction Job"。

• Dedicated Compaction Job

默认情况下，Paimon writer 在写入记录时会根据需要执行Compaction。这对于大多数用例来说已经足够了，但有两个缺点：

这可能会导致写入吞吐量不稳定，因为执行压缩时吞吐量可能会暂时下降。

Compaction会将某些数据文件标记为“已删除”（并未真正删除）。如果多个writer标记同一个文件，则在提交更改时会发生冲突。Paimon 会自动解决冲突，但这可能会导致作业重新启动。

为了避免这些缺点，用户还可以选择在writer中跳过Compaction，并仅运行专门的作业来进行Compaction。由于Compaction仅由专用作业执行，因此writer可以连续写入记录而无需暂停，并且不会发生冲突。

Flink SQL目前不支持compaction相关的语句，所以我们必须通过flink run来提交compaction作业。

<FLINK_HOME>/bin/flink run \
/path/to/paimon-flink-action-0.5-SNAPSHOT.jar \
compact \
–warehouse \
–database \
–table \
[–partition ] \
[–catalog-conf [–catalog-conf …]] \

如果提交一个批处理作业（execution.runtime-mode:batch），当前所有的表文件都会被Compaction。如果您提交一个流作业（execution.runtime-mode: Streaming），该作业将持续监视表的新更改并根据需要执行Compaction。

2.2.4 表管理

• 管理快照

1）快照过期

Paimon Writer每次提交都会生成一个或两个快照。每个快照可能会添加一些新的数据文件或将一些旧的数据文件标记为已删除。然而，标记的数据文件并没有真正被删除，因为Paimon还支持时间旅行到更早的快照。它们仅在快照过期时被删除。

目前，Paimon Writer在提交新更改时会自动执行过期操作。通过使旧快照过期，可以删除不再使用的旧数据文件和元数据文件，以释放磁盘空间。

设置以下表属性：

注意，保留时间太短或保留数量太少可能会导致如下问题：

批量查询找不到该文件。例如，表比较大，批量查询需要10分钟才能读取，但是10分钟前的快照过期了，此时批量查询会读取到已删除的快照。

表文件上的流式读取作业（没有外部日志系统）无法重新启动。当作业重新启动时，它记录的快照可能已过期。（可以使用Consumer Id来保护快照过期的小保留时间内的流式读取）。

2）回滚快照

<FLINK_HOME>/bin/flink run \
/path/to/paimon-flink-action-0.5-SNAPSHOT.jar \
rollback-to \
–warehouse \
–database \
–table \
–snapshot \
[–catalog-conf [–catalog-conf …]]

• 管理分区

创建分区表时可以设置partition.expiration-time。Paimon会定期检查分区的状态，并根据时间删除过期的分区。

判断分区是否过期：将分区中提取的时间与当前时间进行比较，看生存时间是否超过partition.expiration-time。比如：

CREATE TABLE T (…) PARTITIONED BY (dt) WITH (
'partition.expiration-time' = '7 d',
'partition.expiration-check-interval' = '1 d',
'partition.timestamp-formatter' = 'yyyyMMdd'
);

• 管理小文件

小文件可能会导致：

稳定性问题：HDFS中小文件过多，NameNode会承受过大的压力。
成本问题：HDFS中的小文件会暂时使用最小1个Block的大小，例如128MB。
查询效率：小文件过多查询效率会受到影响。

1）Flink Checkpoint的影响

使用Flink Writer，每个checkpoint会生成 1-2 个快照，并且checkpoint会强制在 DFS 上生成文件，因此checkpoint间隔越小，会生成越多的小文件。

默认情况下，不仅checkpoint会导致文件生成，writer的内存（write-buffer-size）耗尽也会将数据flush到DFS并生成相应的文件。可以启用 write-buffer-spillable 在 writer 中生成溢出文件，从而在 DFS 中生成更大的文件。

所以，可以设置如下：

增大checkpoint间隔
增加 write-buffer-size 或启用 write-buffer-spillable

2）快照的影响

Paimon维护文件的多个版本，文件的Compaction和删除是逻辑上的，并没有真正删除文件。文件只有在 Snapshot 过期后才会被真正删除，因此减少文件的第一个方法就是减少 Snapshot 过期的时间。Flink writer 会自动使快照过期。

分区和分桶的影响

表数据会被物理分片到不同的分区，里面有不同的桶，所以如果整体数据量太小，单个桶中至少有一个文件，建议你配置较少的桶数，否则会出现也有很多小文件。

3）主键表LSM的影响

LSM 树将文件组织成Sorted Runs的运行。Sorted Runs由一个或多个数据文件组成，并且每个数据文件恰好属于一个Sorted Runs。

默认情况下，Sorted Runs数取决于 num-sorted-run.compaction-trigger，这意味着一个桶中至少有 5 个文件。如果要减少此数量，可以保留更少的文件，但写入性能可能会受到影响。

4）仅追加表的文件的影响

默认情况下，Append-Only 还会进行自动Compaction以减少小文件的数量

对于分桶的 Append-only 表，为了排序会对bucket内的文件行Compaction，可能会保留更多的小文件。

5）Full-Compaction的影响

主键表是5个文件，但是Append-Only表（桶）可能单个桶里有50个小文件，这是很难接受的。更糟糕的是，不再活动的分区还保留了如此多的小文件。

建议配置Full-Compaction，在Flink写入时配置‘full-compaction.delta-commits’定期进行full-compaction。并且可以确保在写入结束之前分区被full-compaction。

缩放Bucket

1）说明

由于总桶数对性能影响很大，Paimon 允许用户通过 ALTER TABLE 命令调整桶数，并通过 INSERT OVERWRITE 重新组织数据布局，而无需重新创建表/分区。当执行覆盖作业时，框架会自动扫描旧桶号的数据，并根据当前桶号对记录进行哈希处理。

– rescale number of total buckets

ALTER TABLE table_identifier SET (‘bucket’ = ‘…’)

– reorganize data layout of table/partition

INSERT OVERWRITE table_identifier [PARTITION (part_spec)]
SELECT …
FROM table_identifier
[WHERE part_spec]

注意：

ALTER TABLE 仅修改表的元数据，不会重新组织或重新格式化现有数据。重新组织现有数据必须通过INSERT OVERWRITE来实现。

重新缩放桶数不会影响读取和正在运行的写入作业。

一旦存储桶编号更改，任何新安排的 INSERT INTO 作业写入未重新组织的现有表/分区将抛出 TableException ，并显示如下类似异常：

Try to write table/partition … with a new bucket num …,
but the previous bucket num is … Please switch to batch mode,
and perform INSERT OVERWRITE to rescale current data layout first.

对于分区表，不同的分区可以有不同的桶号。例如：

ALTER TABLE my_table SET ('bucket' = '4');
INSERT OVERWRITE my_table PARTITION (dt = '2022-01-01')
SELECT * FROM …;
ALTER TABLE my_table SET ('bucket' = '8');
INSERT OVERWRITE my_table PARTITION (dt = '2022-01-02')
SELECT * FROM …;

在覆盖期间，确保没有其他作业写入同一表/分区。

注意：对于启用日志系统的表（例如Kafka），请重新调整主题的分区以保持一致性。

重新缩放存储桶有助于处理吞吐量的突然峰值。假设有一个每日流式ETL任务来同步交易数据。该表的DDL和管道如下所示。

2）官方示例：

如下是正在跑的一个作业：

– 建表

CREATE TABLE verified_orders (
    trade_order_id BIGINT,
    item_id BIGINT,
    item_price DOUBLE,
    dt STRING,
PRIMARY KEY (dt, trade_order_id, item_id) NOT ENFORCED
) PARTITIONED BY (dt)
WITH (
'bucket' = '16'
);

– kafka表

CREATE temporary TABLE raw_orders(
    trade_order_id BIGINT,
    item_id BIGINT,
    item_price BIGINT,
    gmt_create STRING,
    order_status STRING
) WITH (
'connector' = 'kafka',
'topic' = '…',
'properties.bootstrap.servers' = '…',
'format' = 'csv'
…
);

– 流式插入16个分桶

INSERT INTO verified_orders
SELECT trade_order_id,
  item_id,
  item_price,
  DATE_FORMAT(gmt_create, 'yyyy-MM-dd') AS dt
FROM raw_orders
WHERE order_status = 'verified';

过去几周运行良好。然而，最近数据量增长很快，作业的延迟不断增加。为了提高数据新鲜度，用户可以执行如下操作缩放分桶：

（1）使用保存点暂停流作业

$ ./bin/flink stop \
–savepointPath /tmp/flink-savepoints \
$JOB_ID

（2）增加桶数

ALTER TABLE verified_orders SET ('bucket' = '32');

（3）切换到批处理模式并覆盖流作业正在写入的当前分区

SET 'execution.runtime-mode' = 'batch';

– 假设今天是2022-06-22

– 情况1：没有更新历史分区的延迟事件，因此覆盖今天的分区就足够了

INSERT OVERWRITE verified_orders PARTITION (dt = '2022-06-22')
SELECT trade_order_id,
  item_id,
  item_price
FROM verified_orders
WHERE dt = '2022-06-22';

• 情况2：有更新历史分区的延迟事件，但范围不超过3天

INSERT OVERWRITE verified_orders
SELECT trade_order_id,
  item_id,
  item_price,
  dt
FROM verified_orders
WHERE dt IN ('2022-06-20', '2022-06-21', '2022-06-22');

（4）覆盖作业完成后，切换回流模式，从保存点恢复（可以增加并行度=新bucket数量）。

SET 'execution.runtime-mode' = 'streaming';
SET 'execution.savepoint.path' = ;
INSERT INTO verified_orders
SELECT trade_order_id,
      item_id,
      item_price,
      DATE_FORMAT(gmt_create, 'yyyy-MM-dd') AS dt
FROM raw_orders
WHERE order_status = 'verified';

以上就是本次分享的内容，谢谢大家。