Apache Paimon官网学习记录_paimon 函数管理

Apache Paimon 官网

paimon 官网学习记录，了解 paimon 是用来设计做什么，有什么优势，使用等

Paimon

特点

支持批流一体，主要用来做实时数据湖

Paimon 并使用 LSM 树结构来支持大量数据更新和高性能查询

和Flink CDC整合，实时入湖更简单

支持 mysql 等直接同步到 paimon，且支持表结构变更

Paimon 的变更日志还可以存储到外部日志系统（例如 Kafka）中或从外部日志系统（例如 Kafka）中消费

支持 hive presto spark flink 等多种计算引擎

支持hdfs oss s3等多种储存

基本概念和存储特点

Basic Concepts | Apache Paimon

File Layouts | Apache Paimon

整体看下来，很多地方和 icerberg 有相似之处，但是在 primary 表功能更强，且支持flink纯流式读取，和lookup join

借鉴了iceberg的文件数：File Layouts | Apache Paimon

学过 iceberg 这个就很熟悉了，不过有一些点需要注意

Bucket是读写的最小存储单元，每个Bucket目录中包含一棵LSM树。

Paimon 表只有一个桶，这意味着它只提供单并行读写

paimon 为什么有delete文件？？？

Table

Table 的一些基本参数

CREATE TABLE MyTable (
    pk BIGINT PRIMARY KEY NOT ENFORCED,
    f1 DOUBLE,
    f2 BIGINT,
    ts TIMESTAMP
) WITH (
-- 指定顺序字段，避免乱序
   'sequence.field' = 'ts',
--设置默认值
    'fields.f2.default-value'='0' ,
-- 合并引擎，仅限于 Primary Key Table
    'merge-engine'='deduplicate',
 
);

主键表 Primary Key Table

主键表是 Paimon 作为流式数据湖的核心，它可以接收上游来自数据库 CDC 或者 Flink Streaming 产生的 Changelog，里面包含 INSERT、UPDATE、DELTE 的数据。

主键由一组包含每个记录的唯一值的列组成。Paimon 通过对每个存储桶内的主键进行排序来强制数据排序，允许用户通过对主键应用过滤条件来实现高性能

注意：table.exec.sink.upsert-materialize 建议始终设置为NONE，当输入乱序时，我们建议您使用 序列字段来纠正乱序。

Merge Engines 

deduplicate

默认的合并引擎，使用主键的最新一条记录作为当前数据

partial-update：

部分更新

注意：

stream 查询时，partial-update必须和 lookup or full-compaction changelog producer 一起使用

默认是不支持删除数据的

支持聚合（Aggregation 模式的所有函数）

这两个是什么意思

sequence-group解决多流更新时，不同流字段的乱序问题

 'fields.g_1.sequence-group'='a,b',
 'fields.g_2.sequence-group'='c,d'

Aggregation

默认是使用last_non_null_value聚合，支持以下

• sum：sum 函数聚合多行中的值。它支持 DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT 和 DOUBLE 数据类型。
• product：乘积函数可以跨多行计算乘积值。它支持 DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT 和 DOUBLE 数据类型。
• count：count 函数对多行中的值进行计数。它支持 INTEGER、BIGINT 数据类型。
• max：max 函数识别并保留最大值。它支持 CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP 和 TIMESTAMP_LTZ 数据类型。
• min：min 函数识别并保留最小值。它支持 CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP 和 TIMESTAMP_LTZ 数据类型。
• last_value：last_value 函数用最近导入的值替换以前的值。它支持所有数据类型。
• last_non_null_value：last_non_null_value 函数用最新的非空值替换以前的值。它支持所有数据类型。
• listagg：listagg 函数将多个字符串值连接成一个字符串。它支持 STRING 数据类型。
• bool_and：bool_and 函数评估布尔集中的所有值是否都为 true。它支持 BOOLEAN 数据类型。
• bool_or：bool_or 函数检查布尔集中是否至少有一个值为 true。它支持 BOOLEAN 数据类型。
• first_value：first_value 函数从数据集中检索第一个空值。它支持所有数据类型。
• first_non_null_value：first_non_null_value 函数选择数据集中的第一个非空值。它支持所有数据类型。
• nested_update：nested_update函数将多行收集到一个数组中（所谓的“嵌套表”）。支持数组数据类型。

这个还有些复杂捏，后面再看看

• collect：collect 函数将元素收集到数组中。您可以设置fields.<field-name>.distinct=true去重元素。它仅支持 ARRAY 类型。
• merge_map：merge_map函数合并输入地图。它仅支持 MAP 类型。

只有 sum, product and count 支持retraction (UPDATE_BEFORE and DELETE)

如果您允许某些功能忽略撤回消息，您可以配置： 'fields.${field_name}.ignore-retract'='true'

注意：流读下，必须 lookup or full-compaction changelog producer，（‘input’ changelog producer is also supported, but only returns input records.）

first-row

保留同一主键的第一行，与 deduplicate合并引擎不同的是，在first-row合并引擎中，it will generate insert only changelog

1. first-row merge engine must be used together with lookup changelog producer.
2. You can not specify sequence.field.
3. Not accept DELETE and UPDATE_BEFORE message. You can config first-row.ignore-delete to ignore these two kinds records.

Changelog Producers

paimon 支持对外提供changelog（比如flink stream read）

none 

需要注意：

对读取paimon的source端可能有压力

另外paimon会合并数据，拿不到完整的changelog

但是大多数场景不需要完整的历史changelog，一般都是选默认的none

 input

上游如果是changelog日志，所有输入记录将保存在单独的changelog文件（和lsm树不影响）

lookup

Lookup应该是为了在上游没有changelog时在commit前查找历史数据生成changelog

Paimon will generate changelog through 'lookup' before co

Full Compaction 

Full Compaction 解耦数据写入和changelog生成，Paimon will compare the results between full compactions and produce the differences as changelog。The latency（延迟） of changelog is affected（影响） by the frequency of full compactions.

延迟可能很高

Sequence and Rowkind 

sequence.field 解决数据乱序问题，sequence.field无论输入顺序如何，具有最大值的记录将是最后合并的记录。

Row Kind Field 

这个怎么使用的？？？

追加表 Append Only Table

如果表没有定义主键，则它是 Append 表，你只能插入 INSERT 的数据。Append 表的 Compaction 只是用来合并小文件。

目前 Append 表有两种模式：Scalable 表 与 Queue 表。推荐使用 Scalable 表，它更简单易懂。

Append Scalable 表

CREATE TABLE MyTable (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING
) WITH (
    'bucket' = '-1'
)

当你定义 bucket 为 -1，且没有主键时，你可以认为它就是一张增强的 Hive 表，它将没有桶的概念 (虽然这种模式把数据放到 bucket-0 目录中，但是所有的读写并没有并发限制，桶被忽略了)，它就是一个普通的数仓表，支持批写批读，支持流写流读，只是它的流读会有一部分乱序 (并不是完全的输入顺序)。

它有如下应用场景：

1. 简单替换 Hive 表，在 Hive 表的基础上拥有湖存储的 ACID 特性。

2. 你可以流写流读它，它有着非常高的吞吐，非常好的易用性。

3. 你也可以使用对它进行批的排序，结合 z-order 等手段，它可以提供非常快速的点查范围查询。

如果你想在传统 TPC-DS 中测试 Paimon 表，请使用此模式。

执行拓扑：

图的上面部分是 Source 数据写入表，可以看到，中间并没有 Shuffler 了，它可以有着非常高的吞吐。

图的下面部分是 Compaction 拓扑，它会监控文件，进行必要性的小文件合并，它是完全非阻塞，纯异步的模式，并不会阻塞正常数据的写。如果你希望关闭它，请配置 'write-only'。

我们的建议是：在一个流写任务中保持 Compaction 拓扑的打开 (或者启动 Dedicated Compaction 作业)，其它写作业统统 write-only，这是非常方便的玩法，Compaction 拓扑不但会合并小文件，还会清理过期 Snapshots 等等。

Append Queue 表

 
CREATE TABLE MyTable (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING
) WITH (
    'bucket' = '10',
    'bucket-key' = 'user_id'
)

在这种模式下，你可以将 Append 表视为一个由 bucket 分隔的队列。同一存储桶中的每条记录都是严格排序的，流式读取会将记录准确地按写入顺序传输到下游。你需要还可以定义 bucket 个数和 bucket-key，以实现更大的并行性 (默认 bucket 为 1)。

目前 Queue 表只能通过设置 bucket-key 的方式，Flink Kafka 默认模式 (Fixed, each Flink task ends up in at most one Kafka partition) 暂时没被支持，但是这在 Roadmap 上。

整张表就像 Kafka 的 Queue 一样在工作，它与 Kafka 的优劣如下：

1. 延时分钟级，而 Kafka 可以是秒级，但是 Paimon 的成本要低得多。

2. 数据沉淀下来，可被计算引擎 AD-HOC 查询。

3. 流读可以进行 Projection & Filter Pushdown，这可以大大降低成本。

另外，此模式的 Compaction 只会进行 bucket 内的小文件合并，所以会造成更多的小文件，你可以配置 'compaction.max.file-num' (默认 50) 更小的此参数来更多的合并小文件。

另外，并不仅仅是顺序，Paimon 的此模式还可以让你定义 Watermark，并且支持最新的 Watermark 对齐的策略。

引擎

使用上和其他数据湖差不多，一般也是外部catalog管理元数据，另外支持历史快照（批处理模式）或者从最新的偏移量（在流模式下）读取等多种方式

Spark 

spark 尽管也可以流式读写paimon，但是基本上就是当做了hive表使用

insert into paimon_catalog.user_log .....
select * from paimon_catalog.user_log 

支持update 和 merge into 语法

1. 主键表
2. 只支持 deduplicate or partial-update 两种MergeEngine （first row 和 agg不支持）
3. 不支持修改主键 

UPDATE my_table SET v = 'new_value' WHERE id = 1;
 
-- 尽管官方没说 MERGE INTO 不支持 first row 和 agg
MERGE INTO target
USING source
ON target.a = source.a
WHEN MATCHED THEN
UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *

Ps： UPDATE 和  MERGE INTO 是spark 自己读取数据合并在写入，还是提交给paimon处理（paimon没有单独的计算能力，所以只能是通过文件解决），怎么做的？

模式演化和Spark Procedure我没看太明白

Flink 

paimon支持flink所有数据类型

 查询时指定分区和主键过滤器，且为主键最左边的前缀指定范围过滤器，查询效率会好很多

-- paimon requires checkpoint interval in streaming mode
SET 'execution.checkpointing.interval' = '10 s';
-- write streaming data to dynamic table
INSERT INTO word_count SELECT word, COUNT(*) FROM user_log GROUP BY word;

SET 'execution.runtime-mode' = 'streaming';
SET 'execution.runtime-mode' = 'batch';
 
-- track the changes of table and calculate the count interval statistics
SELECT `interval`, COUNT(*) AS interval_cnt FROM
    (SELECT cnt / 10000 AS `interval` FROM user_log ) GROUP BY `interval`;

冲突是指？？？

Flink 托管内存 我也没看太懂

paimon 支持 looup join，并且重写了 flink 的looup join

目前来看性能还是可以的，比habse 稍微差一些