0
点赞
收藏
分享

微信扫一扫

ETL的数据挖掘方式

流沙雨帘 03-12 07:30 阅读 1
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 <= 60
    animal <> '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-table
  USING source_table | source-expr AS source-alias
  ON merge-condition
  WHEN MATCHED [AND matched-condition]
    THEN UPDATE SET xxx
  WHEN MATCHED [AND matched-condition]
    THEN DELETE
  WHEN NOT MATCHED [AND not_matched_condition]
    THEN INSERT VALUES (xxx)
  WHEN NOT MATCHED BY SOURCE [AND not-matched-by-source-condition]
    THEN UPDATE SET xxx
  WHEN 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
举报

相关推荐

0 条评论