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, ) 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 overwriteINSERT 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 overwriteSET 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

    -- SyntaxINSERT 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 definitionCREATE TABLE MyTable (    k0 INT,    k1 INT,    v STRING) PARTITIONED BY (k0, k1);-- you can useINSERT OVERWRITE MyTable /*+ OPTIONS('dynamic-partition-overwrite'='false') */ PARTITION (k0 = 0) SELECT k1, v FROM MyTable WHERE false;-- orINSERT OVERWRITE MyTable /*+ OPTIONS('dynamic-partition-overwrite'='false') */ PARTITION (k0 = 0, k1 = 0) SELECT v FROM MyTable WHERE false;

    Flink Job

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

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

    查看drop partition的帮助信息

    /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。

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

      Spark引擎

      要启用更新，需要以下配置：

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

      Spark 支持更新原始类型和结构体类型，例如：

      -- SyntaxUPDATE table_identifier SET column1 = value1, column2 = value2, ... WHERE condition;CREATE TABLE T (  id INT,   s STRUCT,   name STRING)TBLPROPERTIES (  'primary-key' = 'id',   'merge-engine' = 'deduplicate');-- you can useUPDATE 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提交“删除”作业来删除记录。

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

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

      查看删除的帮助信息

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

      Flink1.17+

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

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

        -- SyntaxDELETE FROM table_identifier WHERE conditions;-- The following SQL is an example:-- table definitionCREATE 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 useDELETE 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”作业。

            /bin/flink run \    /path/to/paimon-flink-action-0.7.0-incubating.jar \    merge_into \    --warehouse  \    --database  \    --table  \    [--target_as ] \    --source_table  \    [--source_sql  ...]\    --on  \    --merge_actions  \    --matched_upsert_condition  \    --matched_upsert_set  \    --matched_delete_condition  \    --not_matched_insert_condition  \    --not_matched_insert_values  \    --not_matched_by_source_upsert_condition  \    --not_matched_by_source_upsert_set  \    --not_matched_by_source_delete_condition  \    [--catalog_conf  [--catalog_conf  ...]]    You can pass sqls by '--source_sql  [, --source_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  \    --database  \    --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  
            有关语法使用的解析
             
            https://paimon.apache.org/docs/0.7/how-to/writing-tables/
             
            帮助信息查看：
             
            /bin/flink run \    /path/to/paimon-flink-action-0.7.0-incubating.jar \    merge_into --help