Apache Paimon Flink引擎解析

Paimon 支持 Flink 1.17, 1.16, 1.15 和 1.14，当前 Paimon 提供了两类 Jar 包，一类支持数据读写，另一类支持其它操作（compaction）

Version	      Type	        Jar
Flink 1.18	  Bundled Jar	  paimon-flink-1.18-0.7.0-incubating.jar
Flink 1.17	  Bundled Jar	  paimon-flink-1.17-0.7.0-incubating.jar
Flink 1.16	  Bundled Jar	  paimon-flink-1.16-0.7.0-incubating.jar
Flink 1.15	  Bundled Jar	  paimon-flink-1.15-0.7.0-incubating.jar
Flink 1.14	  Bundled Jar	  paimon-flink-1.14-0.7.0-incubating.jar
Flink Action	Action Jar	  paimon-flink-action-0.7.0-incubating.jar
1
2
3
4
5
6
7

1.环境准备

下载 Flink 后解压

tar -xzf flink-*.tgz
1

拷贝 Paimon bundled jar 包到 Flink 的 lib 目录下

cp paimon-flink-*.jar /lib/
1

拷贝 Hadoop Bundled Jar 包到 Flink 的 lib 目录下

cp flink-shaded-hadoop-2-uber-*.jar /lib/
1

为同时运行多个Flink作业，修改/conf/flink-conf.yaml中的集群配置

taskmanager.numberOfTaskSlots: 2
1

本地启动 Flink 集群

/bin/start-cluster.sh
1

验证 Web UI 查看集群是否已启动并运行

localhost:8081
1

启动Flink SQL客户端来执行SQL脚本

/bin/sql-client.sh
1

2.创建 Paimon Catalog 和 Table

创建 Catalog 和 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
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14

3.使用Flink通用的Catalog创建Table

使用Flink通用的Catalog，需要使用Hive metastore，然后可以使用Paimon、Hive和Flink通用表（Kafka和其他表）中的所有表。

在此模式下，应该使用"connector"选项来创建tables。

Paimon将在hive-site.xml中使用hive.metastore.warehouse.dir，需要使用带有scheme的path，例如，hdfs://....否则，Paimon将使用本地路径。

CREATE CATALOG my_catalog WITH (
    'type'='paimon-generic',
    'hive-conf-dir'='...',
    'hadoop-conf-dir'='...'
);

USE CATALOG my_catalog;

-- create a word count table
CREATE TABLE word_count (
    word STRING PRIMARY KEY NOT ENFORCED,
    cnt BIGINT
) WITH (
    'connector'='paimon'
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

4.创建 Source 数据表

-- 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;
1
2
3
4
5
6
7
8
9
10
11
12
13

5.数据查询案例

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;
1
2
3
4
5
6
7
8
9

流式查询

-- 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`;
1
2
3
4
5
6

6.退出Flink SQL客户端，停止Flink集群

退出Flink SQL客户端

-- uncomment the following line if you want to drop the dynamic table and clear the files
-- DROP TABLE word_count;

-- exit sql-client
EXIT;
1
2
3
4
5

停止Flink集群

./bin/stop-cluster.sh
1

7.触发 Savepoint 和 recover

由于Paimon有自己的snapshot管理，可能与Flink的checkpoint管理相冲突，在从savepoint恢复时会导致异常（不会导致存储损坏）。

建议使用以下方法开启savepoint：

使用Stop with savepoint。
使用 Tag with savepoint，并在从savepoint恢复之前rollback-to-tag。

8.使用Action Jar

Flink本地集群启动后，使用以下命令执行 action jar

/bin/flink run \
 /path/to/paimon-flink-action-0.7.0-incubating.jar \
 
 
1
2
3
4

compact 一张 table

/bin/flink run \
 /path/to/paimon-flink-action-0.7.0-incubating.jar \
 compact \
 --path 
1
2
3
4

9.支持的Flink数据类型

支持所有Flink数据类型，除了

• MULTISET不受支持。
• MAP不支持作为主键。

10.使用Flink Managed Memory

Paimon 任务可以创建 memory pools，基于Flink executor管理的 executor memory , 像Flink任务管理的 managed memory。

通过 executor 管理的多个任务的 writer buffers 可以提升 sinks 的稳定性和性能。

使用 Flink managed memory 的配置如下:

在SQL中为Flink Managed Memory设置内存权重，然后Flink sink operator将获得memory pool大小，并为Paimon writer创建allocator。

INSERT INTO paimon_table /*+ OPTIONS('sink.use-managed-memory-allocator'='true', 'sink.managed.writer-buffer-memory'='256M') */
SELECT * FROM ....;
1
2

11.Setting dynamic options

与Paimon表交互时，可以在不更改catalog options的情况下调整table options。

Paimon将获取job-level的dynamic options，并在current session中生效，dynamic options的格式是：

paimon.${catalogName}.${dbName}.${tableName}.${config_key}

catalogName/dbName/tableName 可以是 *
1
2
3

例如：

-- set scan.timestamp-millis=1697018249000 for the table mycatalog.default.T
SET 'paimon.mycatalog.default.T.scan.timestamp-millis' = '1697018249000';
SELECT * FROM T;

-- set scan.timestamp-millis=1697018249000 for the table default.T in any catalog
SET 'paimon.*.default.T.scan.timestamp-millis' = '1697018249000';
SELECT * FROM T;
1
2
3
4
5
6
7

12.Procedures

Flink 1.18及以上版本支持Call Statements，通过编写SQL来操作Paimon表的数据和元数据。

注意：当 call 一个 procedure 时，必须按顺序传递参数，如果不想传递某些参数，必须使用 ‘’ 作为占位符。

例如，用并行度为4的任务压缩表default.t，但不想指定分区和排序策略，调用语句应该是
CALL sys.compact('default.t', '', '', '', 'sink.parallelism=4')

指定分区：使用字符串来表示partition filter，“,“表示"AND”，”;"表示“OR”。

例如，指定两个分区date=01或date=02，需要写’date=01;date=02’；如果指定一个带有date=01和day=01的分区，需要写’date=01,day=01’。

table options 语法：使用字符串来表示table options，格式是’key1=value1,key2=value2…'。