Apache Paimon系列之：Append Table和Append Queue

一、Append Table

如果表没有定义主键，则默认为追加表。

您只能以流式方式将完整记录插入到表中。此类表适合不需要流式更新的用例（例如日志数据同步）。

CREATE TABLE my_table (product_id BIGINT,price DOUBLE,sales BIGINT
);

二、Data Distribution

默认情况下，append table没有bucket的概念。它的作用就像一个 Hive 表。数据文件放置在分区下，可以在其中重新组织和重新排序以加快查询速度。

三、自动小文件合并

在流式写入作业中，如果没有bucket定义，则writer中不会进行压缩，而是使用Compact Coordinator扫描小文件并将压缩任务传递给Compact Worker。在流模式下，如果在flink中运行insert sql，拓扑将是这样的：

不用担心反压，压实永远不会反压。

如果将 write-only 设置为 true，Compact Coordinator 和 Compact Worker 将在拓扑中删除。

自动压缩仅在 Flink 引擎流模式下支持。您还可以通过 paimon 中的 flink 操作在 flink 中启动压缩作业，并通过 set write-only 禁用所有其他压缩。

四、Append Queue

在这种模式下，您可以将append table视为一个由bucket分隔的队列。同一个桶中的每条记录都是严格排序的，流式读取会严格按照写入的顺序将记录传输到下游。使用此模式，不需要进行特殊配置，所有数据都会以队列的形式放入一个桶中。您还可以定义bucket和bucket-key以实现更大的并行性和分散数据。

CREATE TABLE my_table (product_id BIGINT,price DOUBLE,sales BIGINT
) WITH ('bucket' = '8','bucket-key' = 'product_id'
);

五、压缩

默认情况下，sink节点会自动进行compaction来控制文件数量。以下选项控制压缩策略：

Key	Default	Type	Description
write-only	false	Boolean	如果设置为 true，将跳过压缩和快照过期。此选项与专用紧凑作业一起使用。
compaction.min.file-num	5	Integer	对于文件集 [f_0,…,f_N]，满足 sum(size(f_i)) >= targetFileSize 触发追加表压缩的最小文件号。该值避免了几乎完整的文件被压缩，这是不划算的。
compaction.max.file-num	50	Integer	对于文件集 [f_0,…,f_N]，触发追加表压缩的最大文件数，即使 sum(size(f_i)) < targetFileSize。该值可以避免挂起太多小文件，从而降低性能。
full-compaction.delta-commits	(none)	Integer	增量提交后将不断触发完全压缩。

六、Streaming Source

目前仅 Flink 引擎支持流式源行为。

Streaming Read Order

对于流式读取，记录按以下顺序生成：

• 对于来自两个不同分区的任意两条记录
  • 如果 scan.plan-sort-partition 设置为 true，则首先生成分区值较小的记录。
  • 否则，将先产生分区创建时间较早的记录。
• 对于来自同一分区、同一桶的任意两条记录，将首先产生第一条写入的记录。
• 对于来自同一分区但两个不同桶的任意两条记录，不同的桶由不同的任务处理，它们之间没有顺序保证。

七、Watermark Definition

您可以定义读取 Paimon 表的水印：

CREATE TABLE t (`user` BIGINT,product STRING,order_time TIMESTAMP(3),WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (...);-- launch a bounded streaming job to read paimon_table
SELECT window_start, window_end, COUNT(`user`) FROM TABLE(TUMBLE(TABLE t, DESCRIPTOR(order_time), INTERVAL '10' MINUTES)) GROUP BY window_start, window_end;

给定的代码创建了一个名为"t"的表，它包含了三个列：“user”（BIGINT类型），“product”（STRING类型），和"order_time"（TIMESTAMP类型，精确到毫秒）。它还为"order_time"列定义了一个水印（WATERMARK），表示事件的时间戳早于水印的事件被认为是延迟事件，可以被丢弃。

在创建表之后，该代码启动了一个有界的流作业来读取"t"表中的数据。它使用TUMBLE函数将数据按照"order_time"列分组为固定大小的滚动窗口，窗口大小为10分钟。window_start和window_end表示每个窗口的起始和结束时间戳，COUNT函数用于计算每个窗口内不同用户的数量。结果按照window_start和window_end进行分组。

您还可以启用 Flink Watermark 对齐，这将确保没有源/拆分/分片/分区将其水印增加得远远超出其他部分：

Key	Default	Type	Description
scan.watermark.alignment.group	(none)	String	一组用于对齐水印的源。
scan.watermark.alignment.max-drift	(none)	Duration	在我们暂停从源/任务/分区进行消耗之前，对齐水印的最大漂移。

八、Bounded Stream

Streaming Source 也可以是有界的，您可以指定 scan.bounded.watermark 来定义有界流模式的结束条件，流读取将结束，直到遇到更大的水印快照。

快照中的水印是由writer生成的，例如，您可以指定kafka源并声明水印的定义。当使用此kafka源写入Paimon表时，Paimon表的快照将生成相应的水印，以便您在流式读取此Paimon表时可以使用有界水印的功能。

CREATE TABLE kafka_table (`user` BIGINT,product STRING,order_time TIMESTAMP(3),WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka'...);-- launch a streaming insert job
INSERT INTO paimon_table SELECT * FROM kakfa_table;-- launch a bounded streaming job to read paimon_table
SELECT * FROM paimon_table /*+ OPTIONS('scan.bounded.watermark'='...') */;

这段代码包含了几个步骤：

• 创建一个名为"kafka_table"的表，该表包含了三个列：“user”（BIGINT类型），“product”（STRING类型），和"order_time"（TIMESTAMP类型，精确到毫秒）。它还为"order_time"列定义了一个水印（WATERMARK），表示事件的时间戳早于水印的事件被认为是延迟事件，可以被丢弃。该表的连接器（connector）被指定为"kafka"，意味着数据将从Kafka中读取。
• 启动一个流式插入作业。这个作业将从"kafka_table"中选择所有的数据，并插入到名为"paimon_table"的表中。
• 启动一个有界的流作业来读取"paimon_table"表中的数据。该作业将返回"paimon_table"表中的所有数据。注释中的"scan.bounded.watermark"选项可以指定有界流作业的水印，用于确定数据的处理范围。
• 总的来说，这段代码创建了一个从Kafka读取数据的表，并通过流式插入将数据插入到另一个表中。然后，通过有界流作业从目标表中读取数据。