Paimon与Flink集成

使用限制

EMR-3.46.0及后续版本、EMR-5.12.0及后续版本的集群，支持Flink SQL对Paimon进行读写操作。

操作步骤

步骤一：启动集群

本文以Session模式为例，其余模式请参见基础使用。

执行以下命令，启动YARN Session。

yarn-session.sh --detached

步骤二：创建Catalog

Paimon将数据和元数据都保存在文件系统（例如，HDFS）或对象存储（例如，OSS）中，存储的根路径由warehouse参数指定。如果指定的warehouse路径不存在，将会自动创建该路径；如果指定的warehouse路径存在，您可以通过该Catalog访问路径中已有的表。

您还可以将元数据额外同步到Hive或DLF中，方便其它服务访问Paimon。

步骤三 ：流作业读写Paimon

执行以下Flink SQL语句，在Catalog中创建一张表，并读写表中的数据。

-- 设置为流作业。
SET 'execution.runtime-mode' = 'streaming';

-- Paimon在流作业中需要设置checkpoint。
SET 'execution.checkpointing.interval' = '10s';

-- 使用之前创建的catalog。
USE CATALOG test_catalog;

-- 创建并使用一个测试database。
CREATE DATABASE test_db;
USE test_db;

-- 用datagen产生随机数据。
CREATE TEMPORARY TABLE datagen_source (
    uuid int,
    kind int,
    price int
) WITH (
    'connector' = 'datagen',
    'fields.kind.min' = '0',
    'fields.kind.max' = '9',
    'rows-per-second' = '10'
);

-- 创建Paimon表。
CREATE TABLE test_tbl (
    uuid int,
    kind int,
    price int,
    PRIMARY KEY (uuid) NOT ENFORCED
);

-- 向Paimon中写入数据。
INSERT INTO test_tbl SELECT * FROM datagen_source;

-- 读取表中的数据。
-- 流式查询作业运行的过程中，上面触发的流式写入作业仍在运行。
-- 您需要保证Flink集群有足够的资源（task slot）同时运行两个作业，否则无法查到数据。
SELECT kind, SUM(price) FROM test_tbl GROUP BY kind;

步骤四：OLAP查询Paimon

执行以下Flink SQL语句，对刚才创建的表进行OLAP查询。

-- 设置为批作业。
RESET 'execution.checkpointing.interval';
SET 'execution.runtime-mode' = 'batch';

-- 使用tableau展示模式，在命令行中直接打出结果。
SET 'sql-client.execution.result-mode' = 'tableau';

-- 对表中数据进行查询。
SELECT kind, SUM(price) FROM test_tbl GROUP BY kind;

步骤五：清理资源

停止作业后，执行以下Flink SQL语句，删除刚才创建的表。

DROP TABLE test_tbl;