Flink+Paimon+StarRocks(Doris)构建实时湖仓OLAP分析

内容概况

单机环境安装配置可以参考上一篇文章：Flink+Paimon+Hadoop+StarRocks(Doris)单机环境安装部署

    本文主要是基于单机/本地的大数据湖仓环境，测试FlinkSQL + Paimon + StarRocks/ Doris实现的实时（分钟级）数据入湖，并通过StarRocks/Doris外表查询Paimon秒级OLAP查询。

    文章还对「Flink Batch读Paimon」和「StarrRocks读Paimon」做了一些性能对比。

能力介绍

当前 StarRocks x Paimon 的能力主要包括：

1. 支持各类存储系统，包括 HDFS 以及对象存储 S3/OSS/OSS-HDFS
2. 支持 HMS 以及阿里云 DLF 元数据管理系统
3. 支持 Paimon 的 Primary Key 和 Append Only 表类型查询
4. 支持 Paimon 系统表的查询，常见例如 Read Optimized 表，snapshots 表等
5. 支持 Paimon 表和其他类型数据湖格式的关联查询
6. 支持 Paimon 表和 StarRocks 内表的关联查询
7. 支持 Data Cache 加速查询
8. 支持基于 Paimon 表构建物化视图实现透明加速，查询改写等
9. 支持 Paimon 表开启 Delete Vector 加速查询

    对于 Primary Key 表类型，Paimon 社区对 Read Optimized 系统表做了完善的性能优化，可以与 Append Only 表一样充分利用 Native reader 的能力，得到直接查询 Paimon 数据的最佳性能。直接查询 Primary Key 表的情况下，若 Primary Key 表里包含没有做 Compaction 的数据，StarRocks 里会通过 JNI 调用 Java 读取这部分内容，性能会有一定的损耗。即使是这种情况，在我们收到用户反馈里，平均还是会有相对 Trino 达到3倍以上的性能提升。

环境介绍

组件版本Flink1.17.2Paimon0.8.2StarRocks3.2.11

Quick Start

1. 流写Paimon

1.1 启动FlinkSQL

./bin/start-cluster.sh 
./bin/sql-client.sh

1.2 创建paimon catalog

CREATE CATALOG my_catalog WITH (
    'type'='paimon',
    'warehouse'='file:/Users/xxx/bigdata_env/paimon_path'
);

SHOW CATALOGS;

USE CATALOG my_catalog;

USE test_paimon_db;

1.3 创建paimon表 & datagen写入

CREATE TABLE IF NOT EXISTS paimon_tbl_partial_updates_test2 (
 uuid bigint,
 name VARCHAR(15),
 country VARCHAR(15),
 region VARCHAR(15),
 address VARCHAR(35),
 gender VARCHAR(5),
 company VARCHAR(25),
 job VARCHAR(25),
 age int,
 ts TIMESTAMP(3),
 dt VARCHAR(10), 
 PRIMARY KEY (dt, uuid) NOT ENFORCED 
) PARTITIONED BY (dt) 
 WITH (
    'merge-engine' = 'partial-update',
    'changelog-producer' = 'full-compaction', 
    'file.format' = 'orc', 
    'scan.mode' = 'compacted-full', 
    'bucket' = '2', 
    'sink.parallelism' = '2', 
    'sequence.field' = 'ts' 
);

show tables;

SET 'execution.runtime-mode' = 'streaming';  -- !

CREATE TEMPORARY TABLE IF NOT EXISTS source_a2 (
 uuid bigint PRIMARY KEY NOT ENFORCED,
 name VARCHAR(15),
 country VARCHAR(15),
 region VARCHAR(15),
 address VARCHAR(35),
 gender VARCHAR(5),
 company VARCHAR(25),
 job VARCHAR(25),
 age int,
 ts TIMESTAMP(3)
) WITH (
 'connector' = 'datagen', 
 'fields.uuid.kind'='sequence',
 'fields.uuid.start'='0', 
 'fields.uuid.end'='100000', 
 'rows-per-second' = '1' 
);

SET 'execution.checkpointing.interval' = '10 s';

insert into paimon_tbl_partial_updates_test2(uuid, name, country, region, address, gender, company, job, age, ts, dt) select uuid, name, country, region, address, gender, company, job, age, ts, date_format(ts,'yyyy-MM-dd') as dt from source_a2;

2. Flink Batch读Paimon

-- 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 paimon_tbl_partial_updates_test2 WHERE dt='2024-10-17';

-- 
select name from paimon_tbl_partial_updates_test2 where dt='2024-10-17' and (country like '%aa%' or region like '%aa%');

在 localhost:8081 的 Flink-UI 可以看到历史执行情况：

3. StarRocks读Paimon外表

SR通过docker进行部署，需要先通过docker启动镜像容器。

# 挂载启动docker 镜像
docker run -v /Users/xxx/bigdata_env/docker_path/data:/home/zzz/data -p 9030:9030 -p 8030:8030 -p 8040:8040 -itd --name quickstart starrocks/allin1-ubuntu:3.2.11

# docker启动MySQL
docker exec -it quickstart \
mysql -P 9030 -h 127.0.0.1 -u root --prompt="StarRocks > "

由于容器内外的文件系统「不互通」，所以需要挂载或者copy文件夹到docker容器中去！！！

# cp外部文件到容器内
docker cp  ~/bigdata_env/paimon_path quickstart:/home/zzz/data/

通过SR（StarRocks）「外表」查询Paimon

-- 创建paimon catalog（一次性）
CREATE EXTERNAL CATALOG paimon_catalog
PROPERTIES
(
"type" = "paimon",
"paimon.catalog.type" = "filesystem",
"paimon.catalog.warehouse" = "file:/home/zzz/data/paimon_path/"
);

-- test 
SELECT * FROM paimon_catalog.test_paimon_db.word_count;

-- query
select name from paimon_catalog.test_paimon_db.paimon_tbl_partial_updates_test2 where dt='2024-10-17' and (country like '%aa%' or region like '%aa%');

OLAP性能对比

• 在单机版，小数据量上，『SR+Paimon OLAP』查询速度相比『FlinkBatch+Paimon』效率**提升88%**；
• 大数据量需要结合集群环境进一步测试。

测试数据量FlinkBatch+Paimon(ms)SR+Paimoon(ms)300063550506406005051540687407026025000100090675801000100768909801201000901000001000110100010091712010001207471401000300总计147321740
后记：

• 单机环境安装配置可以参考上一篇文章：​​​​​​Flink+Paimon+Hadoop+StarRocks(Doris)单机环境安装部署
• Paimon+IDEA参考：新一代数据湖存储技术Apache Paimon入门Demo
• 基于Paimon主键表的多流拼接：Flink+Paimon多流拼接性能优化实战