Apache Paimon 使用之 Writing Tables

2024-03-08 16:44

本文主要是介绍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

这篇关于Apache Paimon 使用之 Writing Tables的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/787725

相关文章

中文分词jieba库的使用与实景应用(一)

知识星球:https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义: 精确模式(默认模式): 全模式: 搜索引擎模式: paddle 模式(基于深度学习的分词模式): 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

Hadoop数据压缩使用介绍

一、压缩原则 (1)运算密集型的Job,少用压缩 (2)IO密集型的Job,多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1)为了支持多种压缩/解压缩算法,Hadoop引入了编码/解码器 2)要在Hadoop中启用压缩,可以配置如下参数

Makefile简明使用教程

文章目录 规则makefile文件的基本语法:加在命令前的特殊符号:.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具,常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件

使用opencv优化图片(画面变清晰)

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化,使其看起来更清晰,同时保持尺寸不变,通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多,主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器:相机传

pdfmake生成pdf的使用

实际项目中有时会有根据填写的表单数据或者其他格式的数据,将数据自动填充到pdf文件中根据固定模板生成pdf文件的需求 文章目录 利用pdfmake生成pdf文件1.下载安装pdfmake第三方包2.封装生成pdf文件的共用配置3.生成pdf文件的文件模板内容4.调用方法生成pdf 利用pdfmake生成pdf文件 1.下载安装pdfmake第三方包 npm i pdfma

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合,内部除了存储元素外,还会存储一个score,存储在zset中的元素会按照score的大小升序排列,不同元素的score可以重复,score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]

git使用的说明总结

Git使用说明 下载安装(下载地址) macOS: Git - Downloading macOS Windows: Git - Downloading Windows Linux/Unix: Git (git-scm.com) 创建新仓库 本地创建新仓库:创建新文件夹,进入文件夹目录,执行指令 git init ,用以创建新的git 克隆仓库 执行指令用以创建一个本地仓库的

【北交大信息所AI-Max2】使用方法

BJTU信息所集群AI_MAX2使用方法 使用的前提是预约到相应的算力卡,拥有登录权限的账号密码,一般为导师组共用一个。 有浏览器、ssh工具就可以。 1.新建集群Terminal 浏览器登陆10.126.62.75 (如果是1集群把75改成66) 交互式开发 执行器选Terminal 密码随便设一个(需记住) 工作空间:私有数据、全部文件 加速器选GeForce_RTX_2080_Ti

【Linux 从基础到进阶】Ansible自动化运维工具使用

Ansible自动化运维工具使用 Ansible 是一款开源的自动化运维工具,采用无代理架构(agentless),基于 SSH 连接进行管理,具有简单易用、灵活强大、可扩展性高等特点。它广泛用于服务器管理、应用部署、配置管理等任务。本文将介绍 Ansible 的安装、基本使用方法及一些实际运维场景中的应用,旨在帮助运维人员快速上手并熟练运用 Ansible。 1. Ansible的核心概念