Realtime Compute for Apache Flink：流式資料湖倉Paimon

類別

詳情

支援類型

源表、維表和結果表，資料攝入目標端

運行模式

流模式和批模式

資料格式

暫不支援

特有監控指標

暫無

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