背景

本文主要实现使用paimon自带的paimon-flink-action工具，同步mysql数据到StarRocks验证。

paimon-flink-action是一个封装了多种高级功能的Apache Paimon的Flink action jar包‌。这个包内部包含了诸如merge into、compact、delete、drop_partition等功能，主要用于支持复杂的数据操作和数据处理任务。

环境

 flink 1.19 
 paimon 0.9 
 flink cdc 3.2.1 
 starrocks3.3.7 存算分离

jar包准备

paimon-flink-1.20-0.9.0.jar
paimon-flink-action-0.9.0.jar
paimon-s3-0.9.0.jar
flink-sql-connector-mysql-cdc-3.2.1.jar

使用paimon-flink-action

file

./bin/flink run \    
./lib/paimon-flink-action-0.9.0.jar \   
 mysql_sync_database \    
 --warehouse /tmp/warehouse1  \    
 --database app_db \    
 --mysql_conf hostname=ip \    
 --mysql_conf username=root \    
 --mysql_conf password=XXX \    
 --mysql_conf database-name=app_db \    
 --table-conf bucket=1

s3

这个s3的配置有点技巧，mysql-cdc内没有说明，我试了下可以catalog_conf配置，经过咨询官方，原来是在flink的配置文件内可以设置。

./bin/flink run \    
./lib/paimon-flink-action-0.9.0.jar \    
mysql_sync_database \    
--warehouse s3://lakehouse-1253767413/paimon \    
--database app_db \    
--mysql_conf hostname=192.168.103.113 \    
--mysql_conf username=root \    
--mysql_conf password=XXX \    
--mysql_conf database-name=app_db \    
--catalog_conf s3.endpoint=cos.ap-guangzhou.myqcloud.com \    
--catalog_conf s3.access-key=XXX \    
--catalog_conf s3.secret-key=XXX \    
--table-conf bucket=1

starrocks 查询paimon

添加 starrocks paimon catalog

CREATE EXTERNAL CATALOG paimonPROPERTIES(    
"type" = "paimon",    
"paimon.catalog.type" = "filesystem",    
"paimon.catalog.warehouse" = "s3://lakehouse-1253767413/paimon",    
"aws.s3.endpoint" = "cos.ap-beijing.myqcloud.com",    
"aws.s3.access_key" = "XXX",    
"aws.s3.secret_key" = "XXX");

验证

查询paimon数据验证 ✅

select * from paimon.app_db.ordersid

price        log1       
4.00        ["1"]2        
100.00        ["2"]

json验证 ✅ 这里我们发现log字段本来在 mysql 内的是json类型，到了paimon则为string字符串类型
增删改同步 ✅ mysql 增删改，可以同步到paimon，从而starrocks进行查询
创建异步物化视图验证 ✅

CREATE MATERIALIZED VIEW paimon_mv_test
COMMENT "paimon-mv-test" 
REFRESH ASYNC START ( '2024-10-01 10:00:00' ) EVERY ( INTERVAL 10 MINUTE ) 
AS 
select * from paimon.app_db.orders