简介
Apache Paimon 是一种湖格式，支持使用 Flink 和 Spark 构建实时湖仓一体架构 用于流式处理和批处理操作。Paimon创新性地将湖格式与LSM（Log-structured merge-tree）相结合 结构，将实时流式更新引入 Lake 架构。
Paimon提供以下核心能力：
实时更新：
主键表支持大规模更新的写入，具有非常高的更新性能，通常通过 Flink Streaming。
支持定义合并引擎，随心所欲地更新记录。删除重复数据以保留最后一行、部分更新、聚合记录或第一行，由您决定。
支持定义 changelog-producer，在合并引擎的更新中生成正确和完整的更改日志，简化您的流分析。
附加数据处理：
追加表（无主键）提供大规模批处理和流式处理功能。自动小文件合并。
支持z-order排序的数据压缩以优化文件布局，使用 minmax 等索引提供基于数据跳过的快速查询。
数据湖功能：
可扩展元数据：支持存储PB级大规模数据集，存储大量分区。
支持 ACID 事务 & 时间旅行 & 模式演进。
版本
文中使用相关技术版本为flink-1.16和paimon-flink-1.16-0.8.1.jar
添加jar
将paimon-flink-1.16-0.8.1.jar添加到FLINK_HOME/lib目录下
修改FLINK配置文件
vim <FLINK_HOME>/conf/flink-conf.yaml
taskmanager.numberOfTaskSlots: 2
启动FLINK
./bin/start-cluster.sh
执行FLINK SQL
./bin/sql-client.sh
Create a Catalog and a Table
-- if you're trying out Paimon in a distributed environment,
-- the warehouse path should be set to a shared file system, such as HDFS or OSS
CREATE CATALOG my_catalog WITH (
'type'='paimon',
'warehouse'='file:/tmp/paimon'
);
USE CATALOG my_catalog;
-- create a word count table
CREATE TABLE word_count (
word STRING PRIMARY KEY NOT ENFORCED,
cnt BIGINT
);
写数据
-- create a word data generator table
CREATE TEMPORARY TABLE word_table (
word STRING
) WITH (
'connector' = 'datagen',
'fields.word.length' = '1'
);
-- 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 word_table GROUP BY word;
OLAP 查询
-- use tableau result mode
SET 'sql-client.execution.result-mode' = 'tableau';
-- switch to batch mode
RESET 'execution.checkpointing.interval';
SET 'execution.runtime-mode' = 'batch';
-- olap query the table
SELECT * FROM word_count;
流式查询
-- switch to streaming mode
SET 'execution.runtime-mode' = 'streaming';
-- track the changes of table and calculate the count interval statistics
SELECT `interval`, COUNT(*) AS interval_cnt FROM
(SELECT cnt / 10000 AS `interval` FROM word_count) GROUP BY `interval`;
停止FLINK
./bin/stop-cluster.sh
