Apache Paimon 使用之 Writing Tables

Writing Tables
1.插入语法
INSERT { INTO | OVERWRITE } table_identifier [ part_spec ] [ column_list ] { value_expr | query };

part_spec:PARTITION ( partition_col_name = partition_col_val [ , … ] )

column_list:(col_name1 [, column_name2, …])

value_expr:VALUES ( { value | NULL } [ , … ] ) [ , ( … ) ]

注意:Flink 目前不支持直接使用 NULL,需要将其转为对应的数据类型,CAST (NULL AS data_type)

a) 将空字段写入非空字段

不能将另一个表的可空列插入一个表的非空列中,假设在表A中有一个主键为key1,主键不能为空,在表B中有一个列键key2,它是可为空的。如果运行sql:

INSERT INTO A key1 SELECT key2 FROM B

异常如下

  • 在spark中:Cannot write nullable values to non-null column ‘key1’.
  • 在flink中:Column ‘key1’ is NOT NULL, however, a null value is being written into it.

可以使用函数“NVL”或“COALESCE”,将可空列转换为非空列来避免出现异常

INSERT INTO A key1 SELECT COALESCE(key2, <non-null expression>) FROM B;
2.通过select插入表
a) 语法
INSERT INTO MyTable SELECT ...

Paimon 支持在 Sink 阶段通过 partition 和 bucket 来 Shuffle 数据。

b) Overwriting

注意:在Spark中如果spark.sql.sources.partitionOverwriteMode被设置为dynamic,为了确保Paimon表的insert overwrite可以正常使用,那么spark.sql.extensions应该被设置为org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions。

c) Overwriting 整张表

对于未分区的表,Paimon支持overwriting整张表。

INSERT OVERWRITE MyTable SELECT ...
d) Overwriting 一个分区

对于分区表,Paimon支持overwriting一个分区。

INSERT OVERWRITE MyTable PARTITION (key1 = value1, key2 = value2, ...) SELECT ...
e) 动态覆盖

Flink 引擎

Flink的默认覆盖模式是动态分区覆盖(Paimon只删除覆盖数据中显示的分区)可以配置dynamic-partition-overwrite,将其更改为静态覆盖。

-- MyTable is a Partitioned Table-- Dynamic overwrite
INSERT OVERWRITE MyTable SELECT ...-- Static overwrite (Overwrite whole table)
INSERT OVERWRITE MyTable /*+ OPTIONS('dynamic-partition-overwrite' = 'false') */ SELECT ...

Spark 引擎

Spark的默认覆盖模式是静态分区覆盖,要启用动态覆盖,需要以下配置:

--conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
-- MyTable is a Partitioned Table-- Static overwrite (Overwrite whole table)
INSERT OVERWRITE MyTable SELECT ...-- Dynamic overwrite
SET spark.sql.sources.partitionOverwriteMode=dynamic;
INSERT OVERWRITE MyTable SELECT ...
3.Truncate tables

Flink 1.17-

使用INSERT OVERWRITE通过插入空值来清除表

INSERT OVERWRITE MyTable /*+ OPTIONS('dynamic-partition-overwrite'='false') */ SELECT * FROM MyTable WHERE false;

Flink 1.18 和 Spark引擎

TRUNCATE TABLE MyTable;
4.清除分区

目前,Paimon支持两种清除分区的方法。

  • 与清除表一样,使用INSERT OVERWRITE通过插入空值来清除分区的数据。
  • 方法#1不支持删除多个分区。如果需要删除多个分区,可以通过flink run提交drop_partition作业。

Flink SQL

-- Syntax
INSERT OVERWRITE MyTable /*+ OPTIONS('dynamic-partition-overwrite'='false') */ 
PARTITION (key1 = value1, key2 = value2, ...) SELECT selectSpec FROM MyTable WHERE false;-- The following SQL is an example:
-- table definition
CREATE TABLE MyTable (k0 INT,k1 INT,v STRING
) PARTITIONED BY (k0, k1);-- you can use
INSERT OVERWRITE MyTable /*+ OPTIONS('dynamic-partition-overwrite'='false') */ 
PARTITION (k0 = 0) SELECT k1, v FROM MyTable WHERE false;-- or
INSERT OVERWRITE MyTable /*+ OPTIONS('dynamic-partition-overwrite'='false') */ 
PARTITION (k0 = 0, k1 = 0) SELECT v FROM MyTable WHERE false;

Flink Job

运行以下命令为表提交drop partition作业。

<FLINK_HOME>/bin/flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \drop_partition \--warehouse <warehouse-path> \--database <database-name> \--table <table-name> \[--partition <partition_spec> [--partition <partition_spec> ...]] \[--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]partition_spec:
key1=value1,key2=value2...

查看drop partition的帮助信息

<FLINK_HOME>/bin/flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \drop_partition --help
5.更新表
  • 只有主键表支持此功能。
  • MergeEngine需要deduplicate或partial-update才能支持此功能。

注意:不支持更新主键。

Flink 引擎

目前,Paimon支持使用Flink 1.17及更高版本中的UPDATE来更新记录,可以在Flink的batch模式下执行UPDATE

-- Syntax
UPDATE table_identifier SET column1 = value1, column2 = value2, ... WHERE condition;-- The following SQL is an example:
-- table definition
CREATE TABLE MyTable (a STRING,b INT,c INT,PRIMARY KEY (a) NOT ENFORCED
) WITH ( 'merge-engine' = 'deduplicate' 
);-- you can use
UPDATE MyTable SET b = 1, c = 2 WHERE a = 'myTable';

Spark引擎

要启用更新,需要以下配置:

--conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions

Spark 支持更新原始类型和结构体类型,例如:

-- Syntax
UPDATE table_identifier SET column1 = value1, column2 = value2, ... WHERE condition;CREATE TABLE T (id INT, s STRUCT<c1: INT, c2: STRING>, name STRING)
TBLPROPERTIES ('primary-key' = 'id', 'merge-engine' = 'deduplicate'
);-- you can use
UPDATE T SET name = 'a_new' WHERE id = 1;
UPDATE T SET s.c2 = 'a_new' WHERE s.c1 = 1;
6.从表中删除数据

Flink1.16-

在Flink 1.16和以前的版本中,Paimon仅支持通过flink run提交“删除”作业来删除记录。

运行以下命令以提交表的“删除”作业。

<FLINK_HOME>/bin/flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \delete \--warehouse <warehouse-path> \--database <database-name> \--table <table-name> \--where <filter_spec> \[--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]filter_spec 等价于 WHERE 条件在SQL的删除语法中. Examples:age >= 18 AND age <= 60animal <> 'cat'id > (SELECT count(*) FROM employee)

查看删除的帮助信息

<FLINK_HOME>/bin/flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \delete --help

Flink1.17+

  • 只有主键表支持此功能。
  • 如果表有主键,MergeEngine需要deduplicate才能支持此功能。

注意:不支持在流模式下从表中删除。

-- Syntax
DELETE FROM table_identifier WHERE conditions;-- The following SQL is an example:
-- table definition
CREATE TABLE MyTable (id BIGINT NOT NULL,currency STRING,rate BIGINT,dt String,PRIMARY KEY (id, dt) NOT ENFORCED
) PARTITIONED BY (dt) WITH ( 'merge-engine' = 'deduplicate' 
);-- you can use
DELETE FROM MyTable WHERE currency = 'UNKNOWN';

Spark引擎

  • 只有主键表支持此功能。
  • 如果表有主键,MergeEngine需要deduplicate才能支持此功能。

要启用删除,需要以下配置

--conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
DELETE FROM MyTable WHERE currency = 'UNKNOWN';
7.Merging into table

Paimon通过flink run提交“merge_into”作业来支持“MERGE INTO”。

重要的表格属性设置

  • 只有主键表支持此功能。
  • 该操作不会产生UPDATE_BEFORE,因此不建议设置’changelog-producer’ = ‘input’。

语法如下

MERGE INTO target-tableUSING source_table | source-expr AS source-aliasON merge-conditionWHEN MATCHED [AND matched-condition]THEN UPDATE SET xxxWHEN MATCHED [AND matched-condition]THEN DELETEWHEN NOT MATCHED [AND not_matched_condition]THEN INSERT VALUES (xxx)WHEN NOT MATCHED BY SOURCE [AND not-matched-by-source-condition]THEN UPDATE SET xxxWHEN NOT MATCHED BY SOURCE [AND not-matched-by-source-condition]THEN DELETE

merge_into操作使用“upsert”语义而不是“update”,如果行存在,则更新,否则插入。

例如,对于非主键表,可以更新每列,但对于主键表,如果想更新主键,则必须插入一个新行,该行的主键与表中的行不同。在这种情况下,“upsert”是有用的。

Flink Job:运行以下命令为表提交“merge_into”作业。

<FLINK_HOME>/bin/flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \merge_into \--warehouse <warehouse-path> \--database <database-name> \--table <target-table> \[--target_as <target-table-alias>] \--source_table <source_table-name> \[--source_sql <sql> ...]\--on <merge-condition> \--merge_actions <matched-upsert,matched-delete,not-matched-insert,not-matched-by-source-upsert,not-matched-by-source-delete> \--matched_upsert_condition <matched-condition> \--matched_upsert_set <upsert-changes> \--matched_delete_condition <matched-condition> \--not_matched_insert_condition <not-matched-condition> \--not_matched_insert_values <insert-values> \--not_matched_by_source_upsert_condition <not-matched-by-source-condition> \--not_matched_by_source_upsert_set <not-matched-upsert-changes> \--not_matched_by_source_delete_condition <not-matched-by-source-condition> \[--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]You can pass sqls by '--source_sql <sql> [, --source_sql <sql> ...]' to config environment and create source table at runtime.-- Examples:
-- Find all orders mentioned in the source table, then mark as important if the price is above 100 
-- or delete if the price is under 10.
./flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \merge_into \--warehouse <warehouse-path> \--database <database-name> \--table T \--source_table S \--on "T.id = S.order_id" \--merge_actions \matched-upsert,matched-delete \--matched_upsert_condition "T.price > 100" \--matched_upsert_set "mark = 'important'" \--matched_delete_condition "T.price < 10" -- For matched order rows, increase the price, and if there is no match, insert the order from the 
-- source table:
./flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \merge_into \--warehouse <warehouse-path> \--database <database-name> \--table T \--source_table S \--on "T.id = S.order_id" \--merge_actions \matched-upsert,not-matched-insert \--matched_upsert_set "price = T.price + 20" \--not_matched_insert_values * -- For not matched by source order rows (which are in the target table and does not match any row in the
-- source table based on the merge-condition), decrease the price or if the mark is 'trivial', delete them:
./flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \merge_into \--warehouse <warehouse-path> \--database <database-name> \--table T \--source_table S \--on "T.id = S.order_id" \--merge_actions \not-matched-by-source-upsert,not-matched-by-source-delete \--not_matched_by_source_upsert_condition "T.mark <> 'trivial'" \--not_matched_by_source_upsert_set "price = T.price - 20" \--not_matched_by_source_delete_condition "T.mark = 'trivial'"-- A --source_sql example: 
-- Create a temporary view S in new catalog and use it as source table
./flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \merge_into \--warehouse <warehouse-path> \--database <database-name> \--table T \--source_sql "CREATE CATALOG test_cat WITH (...)" \--source_sql "CREATE TEMPORARY VIEW test_cat.`default`.S AS SELECT order_id, price, 'important' FROM important_order" \--source_table test_cat.default.S \--on "T.id = S.order_id" \--merge_actions not-matched-insert\--not_matched_insert_values *

有关语法使用的解析

https://paimon.apache.org/docs/0.7/how-to/writing-tables/

帮助信息查看:

<FLINK_HOME>/bin/flink run \/path/to/paimon-flink-action-0.7.0-incubating.jar \merge_into --help

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/730849.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Python学习笔记-Flask实现简单的抽奖程序(增加图片显示)

1.创建static文件夹,存放图片文件 2.hero列表数据更改为要抽奖的图片名 3.html中可以编写python语句,遍历hero列表内容渲染到表格中 4.在点击随机抽取后,可以获得名称,然后使用img标签,将获取的名称拼接到路径中 3.初始页面,访问127.0.0.1:5000/index 4.点击随机抽取后 5.py…

【面试题】mysql常见面试题及答案总结

事务中的ACID原则是什么? Mysql是如何实现或者保障ACID的? ACID原则是数据库事务管理中必须满足的四个基本属性&#xff0c;确保了数据库事务的可靠性和数据完整性。 简写全称解释实现A原子性&#xff08;Atomicity&#xff09;一个事务被视为一个不可分割的操作序列&#…

【深入理解设计模式】享元设计模式

享元设计模式 概述 享元设计模式&#xff08;Flyweight Design Pattern&#xff09;是一种用于性能优化的设计模式&#xff0c;它通过共享尽可能多的相似对象来减少对象的创建&#xff0c;从而降低内存使用和提高性能。享元模式的核心思想是将对象的共享部分提取出来&#xff…

人机交互中的定向、定性、定量

在人机交互中&#xff0c;定向、定性和定量分析都扮演着重要的角色&#xff0c;帮助设计师更好地理解用户需求、优化系统设计&#xff0c;并评估系统的性能和效果。这种综合的分析方法有助于打造更符合用户期望、更高效的人机交互系统。 在人机交互中&#xff0c;定向&#xff…

实用干货:分享4个冷门但非常实用的HTML属性

大家好&#xff0c;我是大澈&#xff01; 本文约1100字&#xff0c;整篇阅读大约需要2分钟。 关注微信公众号&#xff1a;“程序员大澈”&#xff0c;免费加入问答群&#xff0c;一起交流技术难题与未来&#xff01; 现在关注公众号&#xff0c;免费送你 ”前后端入行大礼包…

炉石传说(ccf201609-3)解题思路

题目 * 玩家会控制一些角色&#xff0c;每个角色有自己的生命值和攻击力。当生命值小于等于 0 时&#xff0c;该角色死亡。角色分为英雄和随从。   * 玩家各控制一个英雄&#xff0c;游戏开始时&#xff0c;英雄的生命值为 30&#xff0c;攻击力为 0。当英雄死亡时&#xff…

软件设计师软考题目解析23 --每日五题

想说的话&#xff1a;要准备软考了。0.0&#xff0c;其实我是不想考的&#xff0c;但是吧&#xff0c;由于本人已经学完所有知识了&#xff0c;只是被学校的课程给锁在那里了&#xff0c;不然早找工作去了。寻思着反正也无聊&#xff0c;就考个证玩玩。 本人github地址&#xf…

TensorRT是什么,有什么作用,如何使用

TensorRT 是由 NVIDIA 提供的一个高性能深度学习推理&#xff08;inference&#xff09;引擎。它专为生产环境中的部署而设计&#xff0c;用于提高在 NVIDIA GPU 上运行的深度学习模型的推理速度和效率。以下是关于 TensorRT 的详细介绍&#xff1a; TensorRT 是 NVIDIA 推出的…

Hive中增量插入的处理

增量数据采集&#xff0c;目前实现的方式是hive中按某个字段创建分区表&#xff0c; insert override的时候where语句带上对应的增量过滤条件。 我一般选取日期字段ETL_DATE。 hive建立分区表&#xff0c;hql如下&#xff1a; CREATE TABLE IF NOT EXISTS product_sell( cate…

抖店怎么运营?学会这个,玩赚整个抖店市场!

我是电商珠珠 我做电商已经有五年的时间了&#xff0c;做抖店也3年多了&#xff0c;期间还带着学生一起做店。 今天就来给你们讲讲店铺的运营流程&#xff0c;你只要按照这个流程去做店&#xff0c;理解了其中的精髓&#xff0c;就会有明显的效果。 一、类目 抖店运营的第一…

freeRTOS20240308

1.总结任务的调度算法&#xff0c;把实现代码再写一下 2.总结任务的状态以及是怎么样进行转换的

Java集合面试题(day 02)

&#x1f4d1;前言 本文主要是【JAVA】——Java集合面试题的文章&#xff0c;如果有什么需要改进的地方还请大佬指出⛺️ &#x1f3ac;作者简介&#xff1a;大家好&#xff0c;我是听风与他&#x1f947; ☁️博客首页&#xff1a;CSDN主页听风与他 &#x1f304;每日一句&am…

容器: string

引言: 为什么要有string类型, 就使用字符数组表示字符串不行吗? 原因: 使用字符数组描述文本信息, 无法确定开多大空间, 开多了浪费,开少了不够用使用string封装: 扩容机制:减少了空间的浪费各种接口:方便修改等操作 string的使用 容量相关 size:获取字符个数,不包含\0 (C语言…

从huggingface下载模型像本地加载但是UnicodeDecodeError

我自己是在Linux下出现了这个问题 原文&#xff1a;https://github.com/huggingface/transformers/issues/13674 The path for the AutoModel should be to a directory pointing to a pytorch_model.bin and to a config.json. Since you’re pointing to the .bin file dire…

探究java final、finally、finalize的异同

探究final、finally、finalize的异同 在Java编程语言中&#xff0c;final、finally和finalize是三个看似相似但实际上用途迥异的关键词。它们各自在Java的不同场景中扮演着重要角色。本文旨在深入探讨这三个关键词的含义、用法以及它们之间的区别。 1. final final是一个Jav…

无限debugger的几种处理方式

不少网站会在代码中加入‘debugger’&#xff0c;使你F12时一直卡在debugger&#xff0c;这种措施会让新手朋友束手无策。 js中创建debugger的方式有很多&#xff0c;基础的形式有&#xff1a; ①直接创建debugger debugger; ②通过eval创建debugger&#xff08;在虚拟机中…

ERD Online 快速启动指南:代码下载到首次运行的全流程攻略 ️

&#x1f680; 一、代码下载 ERD online前端代码正常拉取即可&#x1f44c; 后端代码含有子模块&#xff0c;拉取命令如下&#xff1a; git clone --recurse-submodules https://github.com/www-zerocode-net-cn/martin-framework.git &#x1f6e0;️ 二、代码构建 &#x1f3…

PROTEUS可以在单片机设计时帮助你做什么

引言 在单片机&#xff08;MCU&#xff09;设计过程中&#xff0c;验证和调试是非常重要的步骤。然而&#xff0c;使用实际硬件进行验证和调试需要大量的时间和成本。这时&#xff0c;PROTEUS作为一款强大的电子设计自动化软件&#xff0c;可以极大地提高设计效率&#xff0c;…

算法二刷day3

203.移除链表元素 class Solution { public:ListNode* removeElements(ListNode* head, int val) {ListNode *dummyHead new ListNode(0);dummyHead->next head;ListNode *cur dummyHead;while (cur->next ! nullptr) {if (cur->next->val val) {ListNode *tm…

安全防御-第七次

在FW5和FW6之间建立一条IPSEC通道保证10.0.2.0/24网段可以正常访问到192.168.1.0/24 NAT&#xff1a; 安全策略&#xff1a; NAT: 安全策略&#xff1a; 修改服务器映射&#xff1a; 配置IPSEC&#xff1a;