paimon实战 -- 如何使用partial update实现并发写入paimon宽表代替双流join

🕗 发布于 2024-12-13 14:38 paimon Sequence Group partial update

背景

在湖仓业务开发中，进行宽表开发是常见的业务需求，常规方式都是通过双流或者多流进行join实现将多个表的数据合成一条数据写入宽表，但是join的方式会导致flink任务状态很大，任务不好维护。为了解决状态大的问题paimon提供了partial update，可以不使用join，完成多个insert into 写入或更新同一张表的同一条数据，本文将详细介绍partial update的使用方式和注意事项。

Partial Update

建表时指定'merge-engine' = 'partial-update'，那么就会使用部分更新表引擎，可以做到多个 Flink 流任务去更新同一张表，每条流任务只更新一张表的部分列，最终实现一行完整的数据的更新，对于需要拉宽表的业务场景，partial-update 非常适合此场景，而且构建宽表的操作也相对简单。这里所说的多个 Flink 流任务并不是指多个 Flink Job 并发写同一张 Paimon 表，这样需要拆分 Compaction 任务，就不能在每个 Job 的 Writer 端做 Compaction, 需要一个独立的 Compaction 任务，比较麻烦。目前推荐将多条 Flink 流任务 UNION ALL 起来，启动一个 Job 写 Paimon 表。

注意

1.partial-update 必须跟 lookup 或者 full-compaction changelog producer结合使用。Partial无法接收DELETE消息，可以将partial-update.ignore-delete配置为忽略delete消息。

2.对于流读场景，partial-update 表引擎需要结合 Lookup 或者 full-compaction 的 Changelog Producer 一起使用，同时 partial-update 不能接收和处理 DELETE 消息，为了避免接收到 DELETE 消息报错，需要通过配置 'partial-update.ignore-delete' = 'true' 忽略 DELETE 消息。

Sequence Group

Sequence字段并不能解决多流更新的部分更新表的乱序问题，因为多流更新时 Sequence(序列) 字段可能会被另一个流的最新数据覆盖。
因此我们引入了部分更新表的序列组（Sequence Group）机制。它可以解决：

多流更新时出现混乱。每个流定义其自己的序列组。
真正的部分更新，而不仅仅是非空更新。

如下所示：

CREATE TABLE T (
  k INT,
  a INT,
  b INT,
  g_1 INT,
  c INT,
  d INT,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine'='partial-update',
  'fields.g_1.sequence-group'='a,b',
  'fields.g_2.sequence-group'='c,d'
);

INSERT INTO T VALUES (1, 1, 1, 1, 1, 1, 1);

-- g_2 is null, c, d should not be updated
INSERT INTO T VALUES (1, 2, 2, 2, 2, 2, CAST(NULL AS INT));

SELECT * FROM T; -- output 1, 2, 2, 2, 1, 1, 1

-- g_1 is smaller, a, b should not be updated
INSERT INTO T VALUES (1, 3, 3, 1, 3, 3, 3);

SELECT * FROM T; -- output 1, 2, 2, 2, 3, 3, 3

对于 sequence-group，有效的比较数据类型包括：DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP 和 TIMESTAMP_LTZ。

案例

创建Paimon宽表表

CREATE TABLE if not exists MyTable(
id INT ,
col1_1 STRING,
col1_2 STRING,
col1_3 STRING,
col1_4 STRING,
version1 INT ,
col2_1 STRING,
col2_2 STRING,
col2_3 STRING,
col2_4 STRING,
version2 INT,
col3_1 STRING,
col3_2 STRING,
col3_3 STRING,
col3_4 STRING,
version3 INT ,
PRIMARY KEY (id) NOT ENFORCED
)
WITH (
'merge-engine'='partial-update',
'fields.version1.sequence-group'='col1_1,col1_2,col1_3,col1_4',
'fields.version2.sequence-group'='col2_1,col2_2,col2_3,col2_4',
'fields.version3.sequence-group'='col3_1,col3_2,col3_3,col3_4',
'bucket' = '2'
);

DML语句准备
//三个程序分别执行下面三个语句

insert into MyTable
(id,
col1_1,
col1_2,
col1_3,
col1_4,
version1,
col2_1,
col2_2,
col2_3,
col2_4,
version2,
col3_1,
col3_2,
col3_3,
col3_4,
version3)
select
id,
col1_1,
col1_2,
col1_3,
col1_4,
version1,
CAST(NULL AS String)as col2_1,
CAST(NULL AS String)as col2_2,
CAST(NULL AS String)as col2_3,
CAST(NULL AS String)as col2_4,
CAST(NULL AS INT)as version2,
CAST(NULL AS String)as col3_1,
CAST(NULL AS String)as col3_2,
CAST(NULL AS String)as col3_3,
CAST(NULL AS String)as col3_4,
CAST(NULL AS INT)as version3
from
InputTable;

//
insert into MyTable (id  ,
col1_1 ,
col1_2 ,
col1_3 ,
col1_4 ,
version1   ,
col2_1 ,
col2_2 ,
col2_3 ,
col2_4 ,
version2  ,
col3_1 ,
col3_2 ,
col3_3 ,
col3_4 ,
version3)  select 
 id  ,
CAST(NULL AS String) as col1_1 ,
CAST(NULL AS String) as col1_2 ,
CAST(NULL AS String) as col1_3 ,
CAST(NULL AS String) as col1_4 ,
CAST(NULL AS INT) version1   ,
col2_1 ,
col2_2 ,
col2_3 ,
col2_4 ,
 version2  ,
CAST(NULL AS String) as col3_1 ,
CAST(NULL AS String) as col3_2 ,
CAST(NULL AS String) as col3_3 ,
CAST(NULL AS String) as col3_4 ,
CAST(NULL AS INT) as version3  
from 
InputTable;
//
insert into MyTable (id  ,
col1_1 ,
col1_2 ,
col1_3 ,
col1_4 ,
version1   ,
col2_1 ,
col2_2 ,
col2_3 ,
col2_4 ,
version2  ,
col3_1 ,
col3_2 ,
col3_3 ,
col3_4 ,
version3)  select 
id  ,
CAST(NULL AS String) as col1_1 ,
CAST(NULL AS String) as col1_2 ,
CAST(NULL AS String) as col1_3 ,
CAST(NULL AS String) as col1_4 ,
CAST(NULL AS INT) version1   ,
CAST(NULL AS String) as col2_1 ,
CAST(NULL AS String) as col2_2 ,
CAST(NULL AS String) as col2_3 ,
CAST(NULL AS String) as col2_4 ,
CAST(NULL AS INT) as version2  ,
col3_1 ,
col3_2 ,
col3_3 ,
col3_4 ,
version3  
from 
InputTable;

原文地址：https://blog.csdn.net/aA518189/article/details/144435485

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：git SSL certificate problem: unable to get local issuer certificate
下一篇：跨境店铺运营流程

.NET(C#) 如何配置用户首选项及保存用户设置
.NET(C#) 如何配置用户首选项及保存用户设置
阅读更多2024-12-14
【最新】北大数字普惠金融指数数据集-省市县（2011-2023年）
郭峰,王靖一,王芳,孔涛,张勋,程志云.测度中国数字普惠金融发展:指数编制与空间特征[J].经济学(季刊),2020,19(04):1401-1418.时间跨度：省级和城市级指数时间跨度为2011-2
阅读更多2024-12-14
GESP202412 四级【Recamán】题解（AC）
a11ak−1−kkakak−1−kak−1k小杨想知道 Recamán 数列的前n项从小到大排序后的结果。手动计算非常困难，小杨希望你能帮他解决这个问题。
阅读更多2024-12-14
IDEA遇到EasyConnect中的网络资源无法访问的问题
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。原文链接：https://blog.csdn.net/wanshanyu_/article/de
阅读更多2024-12-14
双目摄像头标定方法
此时已经完成标定，左下角为反投影误差，右边为外参可视化。将双目左右目拍的图像上传（左右目最好不少于20张）此时回到主页面，即可看到成功导出。把这些误差大的删除即可。
阅读更多2024-12-14
Servlet、omcat服务器架构与工作原理
Servlet是运行在服务器端的Java程序，它的主要职责之一是接收并处理来自客户端（如浏览器）的HTTP请求。当客户端发送一个请求到服务器时，Servlet可以解析请求中的信息，例如请求的URL路径
阅读更多2024-12-14
Vue生命周期钩子函数：深入解析与实践
作为高级Vue前端开发人员，对Vue组件的生命周期钩子函数有着深刻的理解是至关重要的。生命周期钩子函数是指在Vue组件的创建、更新、销毁等过程中，Vue自动调用的一系列方法。通过这些钩子函数，我们可以
阅读更多2024-12-14
安卓开发--使用android studio发布APP
app发布
阅读更多2024-12-14
数据结构与算法学习笔记----拓扑排序
@ author: 明月清了个风。
阅读更多2024-12-14
python 将数据保存到现有的Excel文件的新工作表
out_file = ‘query.xlsx’df1 = pd.DataFrame(out_data)若直接写入：df1.to_excel(out_file, index=False, sheet_n
阅读更多2024-12-14

paimon实战 -- 如何使用partial update实现并发写入paimon宽表代替双流join

背景

Partial Update

Sequence Group

案例

相关文章