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