当前位置：首页 > 行业知识 > Apache Paimon 使用之 Creating Table

行业知识

Apache Paimon 使用之 Creating Table

2026-11-16 01:46:14 来源：互联网转载

1.创建 Catalog 管理的 Tables

在Paimon Catalog��创建的Tables由Catalog管理，当Tables从Catalog中删除时，其table files也将被删除。

（）

当使用Paimon Catalog，创建一个名为MyTable的managed table，在Catalog的default数据库中有五列，其中dt、hh和user_id是primary keys。

Flink 引擎

（）

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING,    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED);

Spark3引擎

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING) TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id');

Hive引擎

SET hive.metastore.warehouse.dir=warehouse_path;CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING)STORED BY 'org.apache.paimon.hive.PaimonStorageHandler'TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id');

注意：在删除表之前，应停止在表上的插入Job，否则无法完全删除表文件。

I）分区表

创建一个名为MyTable的表，其中dt和hh是分区列，dt、hh和user_id是主键。

Flink引擎

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING,    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED) PARTITIONED BY (dt, hh);

Spark3引擎

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING) PARTITIONED BY (dt, hh) TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id');

Hive引擎

SET hive.metastore.warehouse.dir=warehouse_path;CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING) PARTITIONED BY (     dt STRING,    hh STRING)STORED BY 'org.apache.paimon.hive.PaimonStorageHandler'TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id');

注意：通过配置partition.expiration-time，可以自动删除过期的分区。

ii）选择分区字段

以下三种类型的字段可以定义为数仓中的分区字段：

Creation Time（推荐）：创建时间通常是不可变的，因此可以将其视为分区字段并将其添加到主键中。
Event Time：事件时间是原始表中的一个字段，对于CDC数据，例如从MySQL CDC同步的表或Paimon生成的Changelogs，都是完整的CDC数据，包括UPDATE_BEFORE Records，即使声明了包含主键的分区字段，也可以实现唯一性（需要'changelog-producer'='input'。

CDC op_ts：它不能被定义为分区字段，无法知道以前的record timestamp。

iii）指定统计模式

Paimon将自动收集数据文件的统计数据，以加快查询过程。支持四种模式：

full：收集完整的指标：null_count, min, max。
truncate(length)：长度可以是任何正数，默认模式是truncate(16)这意味着收集空数，最小/最大值，截断长度为16。主要是为了避免过大的列会放大清单文件。
counts：只收集空计数。

none：禁用元数据统计信息收集。

统计收集器模式可以通过'metadata.stats-mode'配置，默认为'truncate(16)'，可以通过设置'fields.{field_name}.stats-mode'来配置字段级别。

iiii）字段默认值

Paimon表目前支持为表属性中的字段设置默认值，请注意，无法指定分区字段和主键字段。

Flink引擎

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING,    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED) PARTITIONED BY (dt, hh)with(    'fields.item_id.default-value'='0');

Spark3引擎

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING) PARTITIONED BY (dt, hh) TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id',    'fields.item_id.default-value'='0');

Hive引擎

SET hive.metastore.warehouse.dir=warehouse_path;CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING)STORED BY 'org.apache.paimon.hive.PaimonStorageHandler'TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id',    'partition'='dt,hh',    'fields.item_id.default-value'='0');

2.Create Table As

表可以通过查询结果创建，例如CREATE TABLE table_b AS SELECT id, name FORM table_a，生成的表 table_b 将等同于创建表并插入带有以下语句的数据：CREATE TABLE table_b (id INT, name STRING); INSERT INTO table_b SELECT id, name FROM table_a;

当使用CREATE TABLE AS SELECT，可以指定主键或分区。

Flink 引擎

/* For streaming mode, you need to enable the checkpoint. */CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT);CREATE TABLE MyTableAs AS SELECT * FROM MyTable;/* partitioned table */CREATE TABLE MyTablePartition (     user_id BIGINT,     item_id BIGINT,     behavior STRING,     dt STRING,     hh STRING) PARTITIONED BY (dt, hh);CREATE TABLE MyTablePartitionAs WITH ('partition' = 'dt') AS SELECT * FROM MyTablePartition;    /* change options */CREATE TABLE MyTableOptions (       user_id BIGINT,       item_id BIGINT) WITH ('file.format' = 'orc');CREATE TABLE MyTableOptionsAs WITH ('file.format' = 'parquet') AS SELECT * FROM MyTableOptions;/* primary key */CREATE TABLE MyTablePk (      user_id BIGINT,      item_id BIGINT,      behavior STRING,      dt STRING,      hh STRING,      PRIMARY KEY (dt, hh, user_id) NOT ENFORCED);CREATE TABLE MyTablePkAs WITH ('primary-key' = 'dt,hh') AS SELECT * FROM MyTablePk;/* primary key + partition */CREATE TABLE MyTableAll (      user_id BIGINT,      item_id BIGINT,      behavior STRING,      dt STRING,      hh STRING,      PRIMARY KEY (dt, hh, user_id) NOT ENFORCED ) PARTITIONED BY (dt, hh);CREATE TABLE MyTableAllAs WITH ('primary-key' = 'dt,hh', 'partition' = 'dt') AS SELECT * FROM MyTableAll;

Spark3引擎

CREATE TABLE MyTable (     user_id BIGINT,     item_id BIGINT);CREATE TABLE MyTableAs AS SELECT * FROM MyTable;/* partitioned table*/CREATE TABLE MyTablePartition (      user_id BIGINT,      item_id BIGINT,      behavior STRING,      dt STRING,      hh STRING) PARTITIONED BY (dt, hh);CREATE TABLE MyTablePartitionAs PARTITIONED BY (dt) AS SELECT * FROM MyTablePartition;/* change TBLPROPERTIES */CREATE TABLE MyTableOptions (       user_id BIGINT,       item_id BIGINT) TBLPROPERTIES ('file.format' = 'orc');CREATE TABLE MyTableOptionsAs TBLPROPERTIES ('file.format' = 'parquet') AS SELECT * FROM MyTableOptions;/* primary key */CREATE TABLE MyTablePk (     user_id BIGINT,     item_id BIGINT,     behavior STRING,     dt STRING,     hh STRING) TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id');CREATE TABLE MyTablePkAs TBLPROPERTIES ('primary-key' = 'dt') AS SELECT * FROM MyTablePk;/* primary key + partition */CREATE TABLE MyTableAll (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING) PARTITIONED BY (dt, hh) TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id');CREATE TABLE MyTableAllAs PARTITIONED BY (dt) TBLPROPERTIES ('primary-key' = 'dt,hh') AS SELECT * FROM MyTableAll;

i）Create Table Like

要创建与另一个表相同的schema、分区和表属性的表，请使用CREATE TABLE LIKE。

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING,    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED);CREATE TABLE MyTableLike LIKE MyTable;-- Create Paimon Table like other connector tableCREATE TABLE MyTableLike WITH ('connector' = 'paimon') LIKE MyTable;

ii）Table Properties

可以指定表属性来启用功能或提高Paimon的性能。

以下SQL创建一个名为MyTable的表，其五列由dt和hh分区，其中dt、hh和user_id是主键，此表有两个属性：'bucket' = '2'和'bucket-key' = 'user_id'。

Flink引擎

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING,    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED) PARTITIONED BY (dt, hh) WITH (    'bucket' = '2',    'bucket-key' = 'user_id');

Spark引擎

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING) PARTITIONED BY (dt, hh) TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id',    'bucket' = '2',    'bucket-key' = 'user_id');

Hive引擎

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING)STORED BY 'org.apache.paimon.hive.PaimonStorageHandler'TBLPROPERTIES (    'primary-key' = 'dt,hh,user_id',    'partition'='dt,hh',    'bucket' = '2',    'bucket-key' = 'user_id');

3.Creating External Tables

外部表被记录，但不由catalog管理，如果外部表被删除，其表文件不会被删除。

Paimon外部表可以在任何catalog中使用，如果不想创建Paimon Catalog，只想读/写Table，可以考虑创建外部表。

注意：如果表已经存在，options 不会像dynamic options一样更新到表的metadata中。

Flink引擎-已弃用，建议使用Paimon Catalog

Flink SQL支持读取和写入外部表，外部Paimon表是通过指定connector和path表属性创建的，以下SQL创建了一个名为MyTable的外部表，有五列，其中表文件的基本路径是hdfs:///path/to/table。

CREATE TABLE MyTable (    user_id BIGINT,    item_id BIGINT,    behavior STRING,    dt STRING,    hh STRING,    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED) WITH (    'connector' = 'paimon',    'path' = 'hdfs:///path/to/table',    'auto-create' = 'true' -- this table property creates table files for an empty table if table path does not exist                           -- currently only supported by Flink);

注意：Flink SQL必须声明所有字段。

Spark引擎

Spark3仅支持通过Scala API创建外部表，以下Scala代码将位于hdfs:///path/to/table的表加载到DataSet。

val dataset = spark.read.format("paimon").load("hdfs:///path/to/table")

Hive引擎

要访问现有的paimon表，可以在Hive中将它们注册为外部表，以下SQL创建了一个名为my_table的外部表，其中表文件的基本路径是hdfs:///path/to/table，由于schema存储在表文件中，所以用户无需定义列。

CREATE EXTERNAL TABLE my_tableSTORED BY 'org.apache.paimon.hive.PaimonStorageHandler'LOCATION 'hdfs:///path/to/table';

4.Creating Temporary Tables

临时表仅支持Flink引擎，与外部表一样，临时表只是记录，但不由当前的Flink SQL session管理。

如果临时表被删除，其resources将不会被删除，当Flink SQL session关闭时，临时表会被丢弃。

如果想将Paimon Catalog与其它表一起使用，但不想将它们存储在其它Catalog中，可以创建一个临时表。

以下Flink SQL创建了一个Paimon Catalog和一个临时表：

CREATE CATALOG my_catalog WITH (    'type' = 'paimon',    'warehouse' = 'hdfs:///path/to/warehouse');USE CATALOG my_catalog;-- Assume that there is already a table named my_table in my_catalogCREATE TEMPORARY TABLE temp_table (    k INT,    v STRING) WITH (    'connector' = 'filesystem',    'path' = 'hdfs:///path/to/temp_table.csv',    'format' = 'csv');SELECT my_table.k, my_table.v, temp_table.v FROM my_table JOIN temp_table ON my_table.k = temp_table.k;

上一篇：QQ公众号运营全攻略，秘籍、案例与实用技巧大揭秘！

下一篇：双服务器电脑怎么设置密码?