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

相关文章

Java中List的contains()方法的使用小结

《Java中List的contains()方法的使用小结》List的contains()方法用于检查列表中是否包含指定的元素,借助equals()方法进行判断,下面就来介绍Java中List的c... 目录详细展开1. 方法签名2. 工作原理3. 使用示例4. 注意事项总结结论:List 的 contain

C#使用SQLite进行大数据量高效处理的代码示例

《C#使用SQLite进行大数据量高效处理的代码示例》在软件开发中,高效处理大数据量是一个常见且具有挑战性的任务,SQLite因其零配置、嵌入式、跨平台的特性,成为许多开发者的首选数据库,本文将深入探... 目录前言准备工作数据实体核心技术批量插入:从乌龟到猎豹的蜕变分页查询:加载百万数据异步处理:拒绝界面

Android中Dialog的使用详解

《Android中Dialog的使用详解》Dialog(对话框)是Android中常用的UI组件,用于临时显示重要信息或获取用户输入,本文给大家介绍Android中Dialog的使用,感兴趣的朋友一起... 目录android中Dialog的使用详解1. 基本Dialog类型1.1 AlertDialog(

Python使用自带的base64库进行base64编码和解码

《Python使用自带的base64库进行base64编码和解码》在Python中,处理数据的编码和解码是数据传输和存储中非常普遍的需求,其中,Base64是一种常用的编码方案,本文我将详细介绍如何使... 目录引言使用python的base64库进行编码和解码编码函数解码函数Base64编码的应用场景注意

使用Sentinel自定义返回和实现区分来源方式

《使用Sentinel自定义返回和实现区分来源方式》:本文主要介绍使用Sentinel自定义返回和实现区分来源方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Sentinel自定义返回和实现区分来源1. 自定义错误返回2. 实现区分来源总结Sentinel自定

Pandas使用SQLite3实战

《Pandas使用SQLite3实战》本文主要介绍了Pandas使用SQLite3实战,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学... 目录1 环境准备2 从 SQLite3VlfrWQzgt 读取数据到 DataFrame基础用法:读

JSON Web Token在登陆中的使用过程

《JSONWebToken在登陆中的使用过程》:本文主要介绍JSONWebToken在登陆中的使用过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录JWT 介绍微服务架构中的 JWT 使用结合微服务网关的 JWT 验证1. 用户登录,生成 JWT2. 自定义过滤

Java中StopWatch的使用示例详解

《Java中StopWatch的使用示例详解》stopWatch是org.springframework.util包下的一个工具类,使用它可直观的输出代码执行耗时,以及执行时间百分比,这篇文章主要介绍... 目录stopWatch 是org.springframework.util 包下的一个工具类,使用它

Java使用Curator进行ZooKeeper操作的详细教程

《Java使用Curator进行ZooKeeper操作的详细教程》ApacheCurator是一个基于ZooKeeper的Java客户端库,它极大地简化了使用ZooKeeper的开发工作,在分布式系统... 目录1、简述2、核心功能2.1 CuratorFramework2.2 Recipes3、示例实践3

springboot security使用jwt认证方式

《springbootsecurity使用jwt认证方式》:本文主要介绍springbootsecurity使用jwt认证方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地... 目录前言代码示例依赖定义mapper定义用户信息的实体beansecurity相关的类提供登录接口测试提供一