Apache Paimon基础记录

基本都是在官网的学习，简单记录一下其中的核心特点

Apache Paimon 官网

Apache Paimon | Apache Paimon

根据官网介绍去快速了解 paimon 是用来设计做什么，可以做什么，对比与其他数据湖有什么特点，如何使用

Paimon

特点

前身是Flink table Store,很多文档和特性都是一样的

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

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

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 通过对每个存储桶内的主键进行排序来强制数据排序，允许用户通过对主键应用过滤条件来实现高性能

Merge Engines

Merge Engines
deduplicate	默认的合并引擎	使用主键的最新一条记录作为当前数据
partial-update	部分更新，null 值不覆盖已有数据	INSERT INTO T VALUES (1, 1,null,null); INSERT INTO T VALUES (1, null,null,1); SELECT * FROM T; -- output 1, 1, null, 1
aggregation	聚合模型（比如doris），需要指定非主键列的聚合类型	CREATE TABLE MyTable ( product_id BIGINT, price DOUBLE, sales BIGINT, PRIMARY KEY (product_id) NOT ENFORCED ) WITH ( 'merge-engine' = 'aggregation', 'fields.price.aggregate-function' = 'max', 'fields.sales.aggregate-function' = 'sum' );
first-row	保留同一主键的第一行，不接受 DELETE 和 UPDATE_BEFORE	与 deduplicate合并引擎不同的是，在 first-row合并引擎中，它将生成仅插入变更日志

Changelog Producers

支持 none input lookup Full Compaction

这个是什么，有何用途？？？？？和flink文档？？

Table Types | Apache Flink Table Store

Primary Key Table | Apache Paimon

追加表 Append Only Table

和 iceberg 类似，目前来看没有什么特殊之处

外部日志 External Log Systems

External Log Systems | Apache Paimon ，但是spark等估计不支持吧

除了底层的表文件之外，Paimon 的变更日志还可以存储到外部日志系统（例如 Kafka）中或从外部日志系统（例如 Kafka）中消费。比如通过指定'log.system' = 'kafka'，用户可以将更改与表文件一起写入 Kafka（应用场景是啥，基本没这样用过啊）

CREATE TABLE T (...)
WITH ('log.system' = 'kafka','kafka.bootstrap.servers' = '...','kafka.topic' = '...'
);

读写demo

使用方式上和iceberg没啥区别，一般还是使用HIVE RDS 来管理元数据

另外支持历史快照（批处理模式）或者从最新的偏移量（在流模式下）读取等多种方式

注册元数据和表

CREATE CATALOG paimon_catalogWITH ('type' = 'paimon','metastore' = 'hive','uri' = 'thrift://<hive-metastore-host-name>:<port>',-- 'hive-conf-dir' = '...', this is recommended in the kerberos environment-- 'hadoop-conf-dir' = '...', this is recommended in the kerberos environment'warehouse' = 'hdfs:///path/to/warehouse'
);USE CATALOG paimon_catalog;CREATE TABLE paimon_catalog.user_log (user_id BIGINT,item_id BIGINT,behavior STRING,dt STRING,hh STRING,PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
)
--需要分区加上 PARTITIONED BY 关键字即可PARTITIONED BY (dt, hh);

spark读取 / 写入

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

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 lookup join + paimon

lookup join 是 flink 的特性，但是结合 paimon 使用性能还不错（比hbase、redis差，mysql没比较过）

流式读取cdc数据写入paimon，既可以做离线查询、olap，又支持flink流式获取数据和lookup join