Apache Paimon | 文件操作
本文将深入探讨各种文件操作对文件产生的影响。文章提供具体的例子和实用技巧，说明 Paimon 是如何有效管理文件的。此外，通过对提交（commit）和压缩（compact）等操作的深入剖析，旨在加深对文件的创建和更新的理解。
先决条件
在深入阅读本文之前，本文假设你已经了解了以下内容：
- Paimon 基本概念；
- Paimon 文件布局；
- 如何在 Flink 中使用 Paimon。
创建目录
通过 ./sql-client.sh 启动 Flink SQL 客户端，并逐一执行以下语句来创建 Paimon 目录：
1 | CREATE CATALOG paimon WITH ( |
执行该创建目录语句会在给定的路径 file:///tmp/paimon 创建一个目录：
1 | $ ls /tmp/paimon |
创建表
执行下面的创建表语句，创建一个有 3 个字段的 Paimon 表：
1 | CREATE TABLE T ( |
执行该创建表语句会在路径 /tmp/paimon/default.db/T 下创建 Paimon 表 T，其模式存储在 /tmp/paimon/default.db/T/schema/schema-0 中。
使用 tree 命令以树状格式列出当前 Catalog 下的内容：
1 | $ tree /tmp/paimon/ |
向表中插入记录
在 Flink SQL 中运行以下插入语句：
1 | INSERT INTO T VALUES (1, 10001, 'varchar00001', '20230501'); |
一旦 Flink 任务完成，记录就会通过一次成功的 commit 写入 Paimon 表。用户可以通过执行查询 SELECT * FROM T 来验证这些记录的可见性，这将返回一行记录：
1 | id a b dt |
提交过程会在路径 /tmp/paimon/default.db/T/snapshot/snapshot-1 下创建一个快照。快照 snapshot-1 的最终文件布局如下所述：
使用 tree 命令以树状格式列出当前 Catalog 下的内容：
1 | $ tree /tmp/paimon/ |
使用 vim 命令查看 snapshot-1 的内容：
1 | vim /tmp/paimon/default.db/T/snapshot/snapshot-1 |
可以看到该快照文件内容包含快照的元数据，如清单和模式 ID：
1 | { |
请注意，清单列表包含快照的所有更改，baseManifestList 是应用 deltaManifestList 中的更改的基础文件。第一次提交会产生 1 个清单文件，并创建 2 个清单列表（文件名称可能与你的实验中的文件名称不同）：
1 | ll /tmp/paimon/default.db/T/manifest |
manifest-401272a3-05f2-4d51-8a3d-27ecee17ca0e-0 是清单文件（manifest 文件），对应上图中的 manifest-1-0，它存储了快照中数据文件的信息。
manifest-list-c996ae59-9dc5-4d67-b691-7f1c20d71526-0 是 baseManifestList（上图中的 manifest-list-1-base），它实际上是空的。
manifest-list-c996ae59-9dc5-4d67-b691-7f1c20d71526-1 是 deltaManifestList（上图中的 manifest-list-1-delta），它包含了对数据文件执行操作的清单条目列表，在本例中为 manifest-1-0。
接下来，我们插入多条记录，这些记录分布在不同的分区中，让我们看看会发生什么。
在 Flink SQL 中，执行以下语句：
1 | INSERT INTO T VALUES |
第二次 commit 后，执行 SELECT * FROM T 将返回 10 条记录：
1 | id a b dt |
第二次 commit 后，同样会创建一个新的快照，即 snapshot-2，使用 ls -atR 查看物理文件布局如下：
1 | $ ls -atR /tmp/paimon/default.db/T |
使用 tree 命令以树状格式列出当前 Catalog 下的内容：
1 | $ tree /tmp/paimon/default.db |
从 snapshot-2 开始，新的文件布局看起来像
从表中删除记录
现在删除满足条件 dt>=20230503 的记录。使用以下命令删除表 T 中 dt>=20230503 的记录：
1 | ./bin/flink run ./lib/paimon-flink-action-0.4.0-incubating.jar delete --path file:///tmp/paimon/default.db/T --where "dt >= '20230503'" |
执行过程输出的信息如下：
1 | ./bin/flink run ./lib/paimon-flink-action-0.4.0-incubating.jar delete --path file:///tmp/paimon/default.db/T --where "dt >= '20230503'" |
以上命令的执行结果，等价于以下 SQL：
1 | DELETE FROM T WHERE dt >= '20230503'; |
笔者当前演示这些操作的时候，Paimon 0.4 尚不支持在 Flink SQL 中直接执行 DELETE FROM T WHERE dt >= '20230503'; 语句进行删除，报错信息如下：
1 | [ERROR] Could not execute SQL statement. Reason: |
具体执行 SQL 的流程：
1 | Flink SQL> CREATE CATALOG paimon WITH ( |
因此，这里使用客户端命令进行删除操作。
另外，还要注意的点是，命令中引用的是 paimon-flink-action-0.4.0-incubating.jar，而不是 paimon-flink-1.17-0.4.0-incubating.jar，请不要混淆了。
关于 delete 操作的更多信息，可以通过以下命令详细了解：
1 | <FLINK_HOME>/bin/flink run /path/to/paimon-flink-action-{{< version >}}.jar delete --help |
如：
1 | ./bin/flink run ./lib/paimon-flink-action-0.4.jar delete --help |
现在继续探讨删除操作对的表文件产生的影响。
第三次 commit 后，创建了一个新的快照 snapshot-3。继续使用 tree 命令来列出表下的文件：
1 | /tmp/paimon |
你会发现，从分区 dt=20230503 到分区 dt=20230510 都没有被删除。相反，在这些分区里创建这了一个新的数据文件，查看分区 dt=20230510 下的信息：
1 | ll /tmp/paimon/default.db/T/dt=20230510/bucket-0 |
这是可以理解的，因为我们在第二次 commit 时插入了一条记录（用 +I[10, 10010, 'varchar00010', '20230510'] 表示），然后在第三次 commit 时删除了这条记录。
执行 SELECT * FROM T 将返回 2 条记录，即：
1 | id a b dt |
从 snapshot-3 开始，新的文件布局看起来类似下图：
请注意，manifest-3-0 包含 8 个 ADD 操作类型的 manifest 条目，对应于 8 个新写入的数据文件。
压缩表
你可能已经注意到，小文件的数量会在连续的快照中增加，这可能会导致读取性能下降。因此，为了减少小文件的数量，需要进行完全压缩。
现在我们触发完全压缩。确保你已经将执行模式设置为批处理（在 flink-conf.yaml 中添加一个条目 execution.runtime-mode: batch，或在执行命令时添加参数 -Dexecution.runtime-mode=BATCH），并通过 flink run 运行一个专门的压缩作业：
1 | <FLINK_HOME>/bin/flink run \ |
本文这里的执行命令是（当前已经在 Flink 的主目录中）：
1 | ./bin/flink run -Dexecution.runtime-mode=BATCH ./lib/paimon-flink-action-0.4.0-incubating.jar compact --path file:///tmp/paimon/default.db/T |
所有当前的表文件将被压缩，并产生一个新的快照，即 snapshot-4，使用 vim 命令打开文件：
1 | vim /tmp/paimon/default.db/T/snapshot/snapshot-4 |
snapshot-4 包含以下信息：
1 | { |
从 snapshot-4 开始，新的文件布局看起来如下
请注意，manifest-4-0 包含 20 个清单条目（18 个 DELETE 操作和 2 个 ADD 操作）。
- 对于分区 dt=20230503 至分区 dt=20230510，对两个数据文件进行两次
DELETE操作； - 对于分区 dt=20230501 至分区 dt=20230502，对同一数据文件进行一次
DELETE操作和一次ADD操作。
更改表
在 Flink SQL 客户端执行下面的语句来配置完全压缩：
1 | ALTER TABLE T SET ('full-compaction.delta-commits' = '1'); |
执行上述语句会为 Paimon 表创建一个新的模式，即 schema-1。但是，在下次 commit 之前，新创建的模式 schema-1 实际上还没有被使用。
快照过期
注意，只有在快照过期且没有消费者依赖快照时，标记的数据文件才会被真正删除。欲了解更多信息，请参见快照过期。
在快照过期过程中，首先确定快照的范围，然后将这些快照内的数据文件标记为删除。仅当存在引用该特定数据文件的类型为 DELETE 的清单条目时，数据文件才被标记为删除。此标记可确保该文件不会被后续快照使用，可以安全地删除。
假设上图中的 4 个快照都要过期了。过期的流程如下：
- 首先删除所有标记的数据文件，并记录任何更改的桶。
- 然后删除任何更新日志文件和关联的清单文件。
- 最后，删除快照本身、写入最早的提示文件。
如果在删除过程中，有任何目录是空的，它们也会被删除。
假设创建了另一个快照，即 snapshot-5，并且触发了快照过期。snapshot-1 到 snapshot-4 将被删除。为简单起见，我们只关注前面的快照中的文件，快照过期后的最终布局如下：
结果，分区 20230503 到 20230510 被物理删除。
Flink流写入
最后，我们将通过使用 CDC 摄取的例子来研究 Flink 流写入。本节将讨论捕获和写入变化数据到 Paimon 的方法，以及异步压缩和快照提交和过期背后的机制。
首先，让我们仔细研究 CDC 数据摄取的工作流程，以及每个相关组件所扮演的独特角色。
MySQL CDC Source统一读取快照数据和增量数据，其中SnapshotReader读取快照数据，BinlogReader读取增量数据。Paimon Sink将数据写入 Paimon 表中，并在桶级别进行归类。其中的CompactManager会异步触发数据压缩。Committer Operator是一个单例（意味着整个流程中只有一个这样的组件），负责提交快照数据和删除过期的快照数据。
接下来，我们介绍一下端到端的数据流。
MySQL Cdc Source 读取快照和增量数据，在归一化（规范化）后将其发出到下游。
Paimon Sink首先在基于堆的LSM树中缓冲新的记录，并在内存缓冲区满时将其刷入磁盘。请注意，每写一个数据文件都是一个排序的运行。在这一点上，没有创建清单文件和快照。就在Flink检查点发生之前，Paimon Sink将刷新所有缓冲区的记录，并向下游发送可提交的消息，在检查点期间由Committer Operator读取并提交。
Paimon Sink 首先在基于堆的 LSM 树中缓冲新的记录，当内存缓冲区满时将其刷新到磁盘。请注意，每个写入的数据文件都是排序段（sorted run）。在这一点上，没有创建清单文件和快照。就在 Flink 检查点发生之前，Paimon Sink 将刷新所有缓冲区的记录，并向下游发送可提交的消息，该消息由 Committer Operator 在检查点期间读取和提交。
在检查点期间，Committer Operator 将创建一个新快照，并将其与清单列表关联，以便该快照包含表内所有数据文件的信息。
之后可能会发生异步压缩，并且 CompactManager 产生的可提交内容包含有关以前文件和合并文件的数据信息，以便 Committer Operator 可以构建相应的清单条目。在这种情况下，Committer Operator 可能在 Flink 检查点期间产生两个快照，一个用于写入的数据（类型为 Append 的快照），另一个用于压缩（类型为 Compact 的快照）。如果在检查点区间内没有写入数据文件，则仅创建类型为 Compact 的快照。Committer Operator 将检查快照的过期情况，并对标记的数据文件进行物理删除。
（END）
All articles on this blog are licensed under CC BY-NC-SA 4.0 unless otherwise stated.
Related Articles
2023-09-06
Apache Paimon 0.5正式发布
Apache Paimon 0.5.0-incubating 版本终于在今天正式发布了。有近 100 位贡献者参与了此版本的开发，共完成了 500 多次代码提交，为社区带来了许多令人兴奋的新功能和改进。 该版本的主要亮点功能如下： CDC 数据入湖的功能已经成熟。 引入了标签的概念，为离线数据仓库提供不可变视图。 主键表的动态桶模式已投入生产。 引入仅追加模式可扩展表，以取代 Hive 表。 CDC摄入Paimon 支持多种方式将数据导入 Paimon 表，并支持模式演化。在 0.5 版本中，新增了许多新功能，包括： MySQL 数据表同步 支持将分片数据同步到一个 Paimon 表中。 支持将所有字段类型映射为字符串。 MySQL 数据库同步 支持合并多个数据库中的多个分片数据。 支持 --mode combined 模式，将所有表整合到一个统一的数据汇中进行同步，并支持在不重启作业的情况下同步新增的表。 从 Kafka 同步数据表 将 Kafka 主题里的数据表同步到 Paimon 表中。 支持 Canal 和 OGG 格式。 从 Kafka...
2023-09-11
Apache Paimon 0.5版本震撼发布：数据导入、查询性能再升级
昨日，Apache Paimon 0.5.0-incubating 版本正式发布。这个版本是在近 100 位贡献者的共同努力下完成的，共提交了 500 多次代码改进，为社区带来了许多令人兴奋的新功能和改进。 其中的亮点之一是 CDC（Change Data Capture）数据入湖的功能已经成熟。在 0.5 版本中，引入了标签的概念，为离线数据仓库提供了不可变视图。此外，主键表的动态桶模式也已经投入生产，并引入了仅追加模式可扩展表来取代 Hive 表。 Paimon 支持多种方式将数据导入 Paimon 表，并且支持模式演化。在 0.5 版本中，新增了许多新功能，例如 MySQL 同步表和数据库、Kafka 同步表和数据库以及 MongoDB 同步集合和数据库。 主键表可以通过在创建表的 DDL（数据定义语言）中指定主键来创建，它接受插入、更新或删除记录操作。而动态桶（Dynamic Bucket）模式是通过将 'bucket' 参数设置为 '-1' 来实现的，Paimon...
2023-06-10
Apache Paimon | 基本概念
快照（Snapshot）快照捕获了一个表在某个时间点上的状态。用户可以通过最新的快照来访问一个表的最新数据。通过时间旅行，用户也可以通过较早的快照访问表的先前状态。 分区（Partition）Paimon 采用了与 Apache Hive 相同的分区概念来分离数据。 分区是一种可选的方式，可以根据特定列的值（如日期、城市和部门等）将表划分为相关部分。每个表可以有一个或多个分区键来标识一个特定的分区。 通过分区，用户可以高效地对表中的某一记录切片进行操作。有关如何将文件划分为多个分区的详细信息，请参见文件布局。 如果定义了主键，分区键必须是主键的一个子集。 桶（Bucket）非分区的表或分区表中的分区会被细分为桶，以便为数据提供额外的结构，这可用于更有效的查询。 一个桶的范围是由记录中的一个或多个列的哈希值决定的。用户可以通过提供 bucket-key 选项来指定桶的列。如果没有指定 bucket-key...
2023-06-10
Apache Paimon | 主键表
Changelog 表是创建表时的默认表类型。用户可以插入、更新或删除表中的记录。 主键由一列或多列组成，其值能唯一地标识表中的每一行记录。Paimon 强制对数据进行排序，这意味着系统将对每个桶内的主键进行排序。利用这个特性，用户可以通过在主键上添加过滤条件来实现高性能。 通过在变更日志表上定义主键，用户可以获得以下功能。 合并引擎当 Paimon 接收器（sink）收到具有相同主键的两条或更多记录时，它将把这些记录合并成一条记录以保持主键的唯一性。通过指定 merge-engine 表属性，用户可以选择如何合并记录。 在 Flink SQL TableConfig 中始终将 table.exec.sink.upsert-materialize 设置为NONE，sink upsert-materialize 可能会导致奇怪的行为。当输入乱序时，建议使用 Sequence Field 进行乱序校正。 去重（Deduplicate）deduplicate 合并引擎是默认的合并引擎。Paimon 只会保留最新的记录，并丢弃具有相同主键的其他记录。 具体来说，如果最新的记录是一条...
2023-06-10
Apache Paimon | 仅追加表（Append Only 表）
如果一个表没有定义主键，那么它默认是一个 append-only 表。 你只能向表中插入完整的记录。不支持删除或更新，也不能定义主键。这种类型的表适用于不需要更新的用例（例如日志数据同步）。 分桶（Bucketing）你也可以为 Append-only 表定义桶的数量，见 Bucket。 建议设置 bucket-key 字段。否则，数据会按照整行进行散列，性能会很差。 压缩（Compaction）默认情况下，接收节点（sink 节点）将自动进行压缩以控制文件数量。以下选项控制压缩策略： Key 默认值 类型 描述 write-only false Boolean 如果设置为 true，压缩和快照过期将被跳过。此选项与专门的压缩作业一起使用。 compaction.min.file-num 5 Integer 对于文件集 [f_0,…,f_N]，满足 sum(size(f_i)) >= targetFileSize...
2023-06-10
Apache Paimon | 外部日志系统
除了底层表文件的变更日志外，Paimon 的变更日志也可以存储在外部日志系统中，比如 Kafka，或者从外部日志系统中消费。通过指定 log.system 表属性，用户可以选择使用哪个外部日志系统。 如果选择使用外部日志系统，那么所有写入表文件中的记录也会写入日志系统。因此，流查询产生的变化会从日志系统读取，而不是表文件。 一致性保证默认情况下，日志系统中的变化要等到快照之后才对消费者可见，就像表文件一样。这种行为保证了精确一次的语义。也就是说，每条记录只被消费者看到一次。 但是，用户也可以指定表的属性 'log.consistency' = 'eventual'，这样写进日志系统的变更日志就可以立即被消费者消费，而不用等待下一个快照。这种行为减少了变更日志的延迟，但由于可能发生的故障，它只能保证至少一次的语义（即，消费者可能会看到重复的记录）。 如果设置 'log.consistency' = 'eventual'，为了获得正确的结果，Flink 中的 Paimon...