Apache Paimon系列之：Append Table和Append Queue
- 一、Append Table
- 二、Data Distribution
- 三、自动小文件合并
- 四、Append Queue
- 五、压缩
- 六、Streaming Source
- 七、Watermark Definition
- 八、Bounded Stream
一、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读取数据的表，并通过流式插入将数据插入到另一个表中。然后，通过有界流作业从目标表中读取数据。
