類別

詳情

支持類型

源表、維表和結果表，數據攝入目標端

運行模式

流模式和批模式

數據格式

暫不支持

特有監控指標

暫無

API種類

SQL，數據攝入YAML作業

是否支持更新或刪除結果表數據

是

參數

說明

數據類型

是否必填

默認值

備注

connector

表類型。

String

否

無

• 如果在Paimon Catalog中創建Paimon表，則無需填寫。

• 如果在其他Catalog中創建Paimon臨時表，則固定值為paimon。

path

表存儲路徑。

String

否

無

• 如果在Paimon Catalog中創建Paimon表，則無需填寫。

• 如果在其他Catalog中創建Paimon臨時表，則為表在HDFS或OSS中的存儲目錄。

auto-create

創建Paimon臨時表時，若指定路徑不存在Paimon表文件，是否自動創建文件。

Boolean

否

false

參數取值如下：

• false（默認）：如果指定路徑不存在Paimon表文件，則報錯。

• true：如果指定路徑不存在，則Flink系統自動創建Paimon表文件。

bucket

每個分區的分桶數。

Integer

否

1

寫入Paimon表的數據將按bucket-key打散至每個bucket中。

bucket-key

分桶關鍵列。

String

否

無

指定將寫入Paimon表的數據按哪些列的值打散至不同的Bucket中。

列名之間用英文逗號（,）分隔，例如'bucket-key' = 'order_id,cust_id'會將數據按order_id列和cust_id列的值進行打散。

changelog-producer

增量數據產生機制。

String

否

none

Paimon可以為任意輸入數據流產生完整的增量數據（所有的update_after數據都有對應的update_before數據），方便下游消費者。增量數據產生機制的可選值如下：

• none（默認值）：不額外產生增量數據。下游仍然可以流讀Paimon表，但讀到的增量數據是不完整的（只有update_after數據，沒有對應的update_before數據）。

• input：將輸入數據流雙寫至增量數據文件中，作為增量數據。

• full-compaction：每次Full Compaction產生完整的增量數據。

• lookup：每次commit snapshot前產生完整的增量數據。

關于增量數據產生機制的選擇，詳情請參見增量數據產生機制。

full-compaction.delta-commits

Full Compaction最大間隔。

Integer

否

無

該參數指定了每commit snapshot多少次之后，一定會進行一次Full Compaction。

lookup.cache-max-memory-size

Paimon維表的內存緩存大小。

String

否

256 MB

該參數值會同時影響維表緩存大小和lookup changelog-producer的緩存大小，兩個機制的緩存大小都由該參數配置。

merge-engine

相同primary key數據的合并機制。

String

否

deduplicate

參數取值如下：

• deduplicate：僅保留最新一條。

• partial-update：用最新數據中非null的列更新相同primary key的現有數據，其它列保持不變。

• aggregation：通過指定聚合函數進行預聚合。

關于數據合并機制的具體分析，詳情請參見數據合并機制。

partial-update.ignore-delete

是否忽略delete（-D）類型的消息。

Boolean

否

false

參數取值如下：

• true：忽略delete消息。

• false：不忽略delete消息。您需要通過sequence.field等配置項來設定Sink對于delete數據的處理策略，否則可能會拋出IllegalStateException或IllegalArgumentException報錯。

ignore-delete

是否忽略delete（-D）類型的消息。

Boolean

否

false

參數取值同partial-update.ignore-delete。

partition.default-name

分區默認名稱。

String

否

__DEFAULT_PARTITION__

如果分區列的值為null或空字符串，將會采用該默認名稱作為分區名。

partition.expiration-check-interval

多久檢查一次分區過期。

String

否

1h

詳情請參見如何設置分區自動過期？

partition.expiration-time

分區的過期時長。

String

否

無

當一個分區的存活時長超過該值時，該分區將會過期，默認永不過期。

一個分區的存活時長由該分區的分區值計算而來，詳情請參見如何設置分區自動過期？

partition.timestamp-formatter

將時間字符串轉換為時間戳的格式串。

String

否

無

設置從分區值提取分區存活時長的格式，詳情請參見如何設置分區自動過期？

partition.timestamp-pattern

將分區值轉換為時間字符串的格式串。

String

否

無

設置從分區值提取分區存活時長的格式，詳情請參見如何設置分區自動過期？

scan.bounded.watermark

當Paimon源表產生的數據的watermark超過該值時，Paimon源表將會結束產生數據。

Long

否

無

無。

scan.mode

指定Paimon源表的消費位點。

String

否

default

詳情請參見如何設置Paimon源表的消費位點？

scan.snapshot-id

指定Paimon源表從哪個snapshot開始消費。

Integer

否

無

詳情請參見如何設置Paimon源表的消費位點？

scan.timestamp-millis

指定Paimon源表從哪個時間點開始消費。

Integer

否

無

詳情請參見如何設置Paimon源表的消費位點？

snapshot.num-retained.max

至多保留幾個最新Snapshot不過期。

Integer

否

2147483647

只要滿足該配置或snapshot.time-retained其中之一，并同時滿足snapshot.num-retained.min，就會觸發Snapshot過期。

snapshot.num-retained.min

至少保留幾個最新Snapshot不過期。

Integer

否

10

無。

snapshot.time-retained

Snapshot產生多久以后會過期。

String

否

1h

只要滿足該配置或snapshot.num-retained.max其中之一，并同時滿足snapshot.num-retained.min，就會觸發snapshot過期。

write-mode

Paimon表的寫入模式。

String

否

change-log

參數取值如下：

• change-log：Paimon表支持根據primary key進行數據的插入、刪除和更新。

• append-only：Paimon表只接受數據的插入，且不支持primary key。該模式比change-log模式更加高效。

關于寫入模式的具體介紹，詳情請參見寫入模式。

scan.infer-parallelism

是否自動推斷Paimon源表的并發度。

Boolean

否

false

參數取值如下：

• true：將會根據分桶數自動推斷Paimon源表的并發度。

• false：按VVP配置的默認并發。如果是專家模式就按用戶配置的并發。

scan.parallelism

Paimon源表的并發度。

Integer

否

無

sink.parallelism

Paimon結果表的并發度。

Integer

否

無

sink.clustering.by-columns

指定寫入Paimon結果表的聚類列。

String

否

無

對于Paimon Append Only表（非主鍵表），在批作業中配置該參數可以啟用聚類寫入功能，使數據在特定列上按大小范圍聚集分布，從而提升該表的查詢速度。

多個列名請使用英文逗號（,）進行分隔，例如'col1,col2'。

聚類詳情請參見Apache Paimon官方文檔。

sink.delete-strategy?

設定校驗策略，確保系統能正確處理回撤（-D/-U）類型消息。

??

Enum

否

NONE

校驗策略取值及Sink算子應當正確處理回撤消息的行為如下：?

• ?NONE（默認值）：不做校驗。?

• IGNORE_DELETE：Sink算子應當忽略-U和-D類型的消息，不發生回撤。

• NON_PK_FIELD_TO_NULL：Sink算子應當忽略-U類型的消息，但是在收到-D類型的消息時，保持主鍵值不變、回撤Schema中其他非主鍵值。

  主要用在多個Sink同時寫入同一張表時部分更新的場景。?

• DELETE_ROW_ON_PK：Sink算子應當忽略-U類型的消息，但是在收到-D類型的消息時刪除主鍵對應的行。?

• CHANGELOG_STANDARD：Sink算子應當在收到-U和-D類型的數據時均會刪除主鍵對應的行。?

合并機制

詳情

去重（Deduplicate）

去重機制（deduplicate）是默認的數據合并機制。對于多條具有相同primary key的數據，Paimon結果表僅會保留最新一條數據，并丟棄其它具有primary key的數據。

部分更新（Partial Update）

通過指定部分更新機制（partial-update），您可以通過多條消息對數據進行逐步更新，并最終得到完整的數據。具體來說，具有相同primary key的新數據將會覆蓋原來的數據，但值為null的列不會進行覆蓋。

例如，假設Paimon結果表按順序收到了以下三條數據：

• <1, 23.0, 10, NULL>

• <1, NULL, NULL, 'This is a book'>

• <1, 25.2, NULL, NULL>

第一列是primary key，則最終結果為<1, 25.2, 10, 'This is a book'>。

預聚合（Aggregation）

部分場景下，可能只關心聚合后的值。預聚合機制（aggregation）將具有相同primary key的數據根據您指定的聚合函數進行聚合。對于不屬于primary key的每一列，都需要通過fields.<field-name>.aggregate-function指定一個聚合函數，否則該列將默認使用last_non_null_value聚合函數。例如，考慮以下Paimon表的定義。

CREATE TABLE MyTable (
  product_id BIGINT,
  price DOUBLE,
  sales BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'merge-engine' = 'aggregation',
  'fields.price.aggregate-function' = 'max',
  'fields.sales.aggregate-function' = 'sum'
);

price列將會根據max函數進行聚合，而sales列將會根據sum函數進行聚合。給定兩條輸入數據 <1, 23.0, 15>和 <1, 30.2, 20>，最終結果為<1, 30.2, 35>。當前支持的聚合函數與對應的數據類型如下：

• sum：支持DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT和DOUBLE。

• min和max：支持DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP和TIMESTAMP_LTZ。

• last_value和last_non_null_value：支持所有數據類型。

• listagg：支持STRING。

• bool_and和bool_or：支持BOOLEAN。