Paimon連接器 - Realtime Compute for Apache Flink

流式資料湖倉Paimon連接器推薦配合Paimon Catalog使用，本文為您介紹如何使用流式資料湖倉Paimon連接器。

背景資訊

Apache Paimon是一種流批統一的湖儲存格式，支援高吞吐的寫入和低延後查詢。目前阿里雲開源巨量資料平台E-MapReduce常見的計算引擎（例如Flink、Spark、Hive或Trino）都與Paimon有著較為完善的整合度。您可以藉助Apache Paimon快速地在HDFS或者雲端OSS上構建自己的資料湖儲存服務，並接入上述計算引擎實現資料湖的分析，詳情請參見Apache Paimon。

類別	詳情
支援類型	源表、維表和結果表，資料攝入目標端
運行模式	流模式和批模式
資料格式	暫不支援
特有監控指標	暫無
API種類	SQL，資料攝入YAML作業
是否支援更新或刪除結果表資料	是

特色功能

目前Apache Paimon提供以下核心能力：

基於HDFS或者Object Storage Service構建低成本的輕量級資料湖儲存服務。
支援在流模式與批模式下讀寫大規模資料集。
支援分鐘級到秒級資料新鮮度的批查詢和OLAP查詢。
支援消費與產生增量資料，可作為傳統的離線數倉和新型的流式數倉的各級儲存。
支援預彙總資料，降低儲存成本與下遊計算壓力。
支援回溯歷史版本的資料。
支援高效的資料過濾。
支援表結構變更。

使用限制與建議

僅Flink計算引擎VVR 6.0.6及以上版本支援Paimon連接器。
Paimon與VVR版本對應關係詳情如下表所示。
Paimon社區版本
Realtime ComputeFlink版引擎版本（VVR ）
1.3.1
11.5、11.6、11.7、11.8
1.3
11.4
1.2
11.2、11.3
1.1
11.1
1.0
8.0.11
並發寫入情境的儲存建議
當多個作業並發更新同一張 Paimon 表時，標準 OSS 儲存（oss://）可能因檔案操作的原子性限制，導致小機率的提交衝突或作業報錯。
為確保持續穩定的寫入，建議使用具備強原子性保障的中繼資料或儲存服務。推薦優先使用資料湖構建（DLF），能提供Paimon中繼資料與儲存服務統一管理，也可選擇 OSS-HDFS 或 HDFS。
參數配置的生效機制
修改Paimon表的配置參數後，必須重啟相關作業，新配置才會生效。運行中的作業無法動態感知或載入此類參數變更。
刪除分區的物理回收延遲
執行 DROP PARTITION 操作時，系統不會立即刪除底層的物理資料檔案。
該操作的本質是邏輯刪除。Paimon 僅在最新產生的快照中移除了目標資料分割的中繼資料。由於 Paimon 具備時間旅行特性，歷史快照仍會保持對該分區資料檔案的引用。只有當引用該分區的所有歷史快照均達到保留期限，並被系統的快照到期機制清理後，物理資料檔案才會被真正刪除。

SQL

Paimon連接器可以在SQL作業中使用，作為源表或者結果表。

文法結構

如果您在Paimon Catalog中建立Paimon表，則無需指定connector參數，此時建立Paimon表的文法結構如下。
```
CREATE TABLE `<YOUR-PAIMON-CATALOG>`.`<YOUR-DB>`.paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  ...
);
```
說明
如果您已在Paimon Catalog中建立了Paimon表，後續無需再次建立表即可直接使用。
如果您在其他Catalog中建立Paimon暫存資料表，則需要指定connector參數與Paimon表的儲存路徑path，此時建立Paimon表的文法結構如下。
```
CREATE TEMPORARY TABLE paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'paimon',
  'path' = '<path-to-paimon-table-files>',
  'auto-create' = 'true', -- 如果指定路徑不存在Paimon表資料檔案，則會自動建立檔案。
  ...
);
```
說明
- 路徑樣本：'path' = 'oss://<bucket>/test/order.db/orders'。不要遺漏 .db 尾碼。Paimon 依賴此命名識別資料庫。
- 多作業寫入一張表，必須共用同一路徑配置。
- 如果兩個path路徑不同，Paimon 就不會認為它們是同一張表。即使實體路徑一樣，只要 Catalog 配置不一致，就會導致並發寫入衝突、compact 失敗、資料丟失等。（如oss://b/testtest與oss://b/test/，多一個斜杠也是不同，但實體路徑相同）

WITH參數

參數	說明	資料類型	是否必填	預設值	備忘
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表檔案。
file.format	表中資料檔案的儲存類型	String	否	parquet	可選值： orc parquet avro lance (VVR 11.6及以上版本） mosaic (VVR 11.9及以上版本)
bucket	每個分區的分桶數。	Integer	否	1	寫入Paimon表的資料將按`bucket-key`打散至每個bucket中。說明建議每個Bucket的資料量在5 GB以下。
bucket-key	分桶關鍵列。	String	否	無	指定將寫入Paimon表的資料按哪些列的值打散至不同的Bucket中。列名之間用英文逗號（,）分隔，例如`'bucket-key' = 'order_id,cust_id'`會將資料按order_id列和cust_id列的值進行打散。說明如果該參數未填寫，則按primary key進行打散。如果Paimon表未指定primary key，則按所有列的值進行打散。
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報錯。說明在Realtime Compute引擎VVR 8.0.6及以下版本，該參數只在partial update情境下，`merge-engine = partial-update`時生效。在Realtime Compute引擎VVR 8.0.7及以上版本，該參數相容適配非partial update情境，與`ignore-delete`參數功能一致，推薦替換成`ignore-delete`。請您根據實際業務情境，判斷出現的delete類型資料是否符合預期，從而決定是否啟用該參數。如果delete類型資料所代表的作業語義不符合預期，則拋出錯誤是更合適的選擇。
ignore-delete	是否忽略delete（-D）類型的訊息。	Boolean	否	false	參數取值同partial-update.ignore-delete。說明僅Realtime Compute引擎VVR 8.0.7及以上版本支援該參數。與partial-update.ignore-delete參數功能一致，推薦使用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	否	true	參數取值如下： 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類型的資料時均會刪除主鍵對應的行。說明僅Realtime Compute引擎VVR 8.0.8及以上版本支援該參數。 Paimon Sink處理回撤訊息的行為實際由ignore-delete、merge-engine等其他配置項的值決定。本配置項不直接影響這部分行為，而是會校正這部分行為是否符合預期策略。在不符合預期策略的情況下，相關校正步驟將終止，並在作業報錯中提示您如何修改ignore-delete、merge-engine等其他配置項以符合預期。
blob-as-descriptor	讀取Blob列時是否輸出Blob描述符位元組。	Boolean	否	false	設為true時，查詢返回BlobDescriptor序列化位元組而非實際Blob內容，配合Blob預簽名URL函數使用。無需建表時指定，可在讀取時通過SQL Hint動態指定。說明僅VVR 11.9-preview1及以上版本支援。

向量表專屬參數

說明

以下參數僅在VVR 11.8及以上版本中支援。

參數	說明	資料類型	是否必填	預設值	備忘
`ivf.nprobe`	搜尋期間探測的IVF聚類數量。	Integer	否	16	數值越高通常召回率越好，但延遲也會增加。
`ivf.refine_factor`	檢索top_k × refine_factor個IVF候選項，並使用Paimon表中儲存的原始向量對其重新排序。	Float	否	禁用（Disabled）	每種IVF變體預設禁用。最適用於壓縮索引（如ivf-pq、ivf-hnsw-sq），適合召回率優先於延遲的情境。
`hnsw.ef_search`	搜尋期間的HNSW搜尋寬度。	Integer	否	0	數值越高通常召回率越好，但延遲也會增加。0表示使用原生庫預設值。
`diskann.search.list_size`	Lumina DiskANN搜尋列表大小。	Integer	否	max(1.5 × top_k, 16)	數值越高通常召回率越好，但延遲也會增加。
`diskann.search.beam_width`	Lumina DiskANN搜尋波束寬度。	Integer	否	4	—
`search.parallel_number`	Lumina搜尋並行數量。	Integer	否	5	—

說明

更多配置項詳情請參見Apache Paimon官方文檔。

特色功能詳解

資料新鮮度與一致性保證

Paimon結果表使用兩階段交易認可協議，在每次Flink作業的checkpoint期間提交寫入的資料，因此資料新鮮度即為Flink作業的checkpoint間隔。每次提交將會產生至多兩個snapshot。

當兩個Flink作業同時寫入一張Paimon表時，如果兩個作業的資料沒有寫入同一個分桶，則能保證serializable層級的一致性。如果兩個作業的資料寫入了同一個分桶，則只能保證snapshot isolation層級的一致性。也就是說，表中的資料可能混合了兩個作業的結果，但不會有資料丟失。

資料合併機制

當Paimon結果表收到多條具有相同primary key的資料時，為了保持primary key的唯一性，Paimon結果表會將這些資料合併成一條資料。通過指定merge-engine參數，您可以指定資料合併的具體行為。資料合併機制詳情如下表所示。

合并機制	詳情
去重（Deduplicate）	去重機制（deduplicate）是預設的資料合併機制。對於多條具有相同primary key的資料，Paimon結果表僅會保留最新一條資料，並丟棄其它具有primary key的資料。說明如果最新一條資料是一條delete訊息，所有具有該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'>。說明如果需要流讀partial-update的結果，必須將`changelog-producer`參數設定為lookup或full-compaction。 partial-update無法處理delete訊息。您可以設定`partial-update.ignore-delete`參數以忽略delete訊息。
預彙總（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。說明只有sum函數支援回撤與刪除資料，其它彙總函式不支援回撤與刪除。如果您需要某些列忽略回撤與刪除訊息，可以設定`'fields.${field_name}.ignore-retract'='true'`。如果需要流讀aggregation的結果，必須將`changelog-producer`參數設定為lookup或full-compaction。

增量資料產生機制

通過changelog-producer參數設定相應的增量資料產生機制，Paimon可以為任意輸入資料流產生完整的增量資料（所有的update_after資料都有對應的update_before資料）。以下列舉了所有的增量資料產生機制，更加詳細的介紹請參見Apache Paimon官方文檔。

機制	詳情
None	設定`changelog-producer`為none（預設值）後，此時，對於同一個primary key，下遊的Paimon源表只能看到資料的最新情況。但這些最新情況無法讓下遊消費者方便地瞭解完整的增量資料，從而進行正確的計算。因為它只能確定對應資料是否被刪除了，或最新資料是什麼，無法得知更改之前的資料是什麼。例如，假設下遊消費者需要計算某一列的總和，如果消費者只看到了最新資料5，它無法斷定該如何更新總和。因為如果之前的資料是4，它應該將總和增加1；如果之前的資料是6，它應該將總和減去1。此類消費者對update_before較為敏感，建議不要將增量資料產生機制配置為None，但是其他增量資料產生機制會帶來效能損耗。說明如果您的下遊是資料庫之類的對update_before資料不敏感的消費者，則可以將增量資料產生機制配置為None。因此，建議您根據實際需要配置增量資料產生機制。
Input	設定`changelog-producer`為input後，Paimon結果表會將輸入資料流雙寫至增量資料檔案中，作為增量資料。因此，只有當輸入資料流本身是完整的增量資料時（例如CDC資料），才能使用這一增量資料產生機制。
Lookup	設定`changelog-producer`為lookup後，Paimon結果表會通過一種類似於維表的點查機制，在每次commit snapshot之前產生本次snapshot對應的完整增量資料。無論輸入資料是否為完整的增量資料，這一增量資料產生機制均能產生完整的增量資料。與下文的Full Compaction機制相比，Lookup機制產生增量資料的時效性更好，但總體來看耗費的資源更多。推薦在對增量資料的新鮮度有較高要求（例如分鐘級）的情況下使用。
Full Compaction	設定`changelog-producer`為full-compaction後，Paimon結果表會在每一次full compaction時產生完整的增量資料。無論輸入資料是否為完整的增量資料，這一增量資料產生機制均能產生完整的增量資料。Full compaction的時間間隔由`full-compaction.delta-commits`參數指定。與上文的Lookup機制相比，Full Compaction機制產生增量資料的時效性更差，但它利用了資料的full compaction過程，不產生額外計算，因此總體來看耗費的資源更少。推薦在對增量資料的新鮮度要求不高（例如小時級）的情況下使用。

寫入模式

Paimon表目前支援的寫入模式如下。

模式

詳情

Change-log

change-log寫入模式是Paimon表的預設寫入模式。該寫入模式支援根據primary key對資料進行插入、刪除與更新，您也可以在該寫入模式下使用上文提到的資料合併機制與增量資料產生機制。

Append-only

append-only寫入模式僅支援資料的插入，且不支援primary key。該模式比change-log模式更加高效，可在對資料新鮮度要求一般的情境下（例如分鐘級新鮮度）作為訊息佇列的替代品。

關於append-only寫入模式的詳細介紹，請參見Apache Paimon官方文檔。在使用append-only寫入模式時，需要注意以下兩點：

建議您根據實際需求設定bucket-key參數，否則Paimon表將根據所有列的值進行分桶，計算效率較低。
append-only寫入模式可在一定程度上保證資料的產出順序，具體的產出順序為：
1. 如果兩條資料來自不同的分區，若設定了scan.plan-sort-partition參數，則分區值較小的資料將首先產出。否則來自較早建立的分區的資料將首先產出。
2. 如果兩條資料來自同一分區的同一分桶，則較早寫入的資料將首先產出。
3. 如果兩條資料來自同一分區的不同分桶，由於不同分桶由不同的並發進行處理，因此不保證兩條資料的產出順序。

作為Flink CDC資料攝入的目標端

Paimon表支援即時同步單表或整庫層級的資料，在同步過程之中如果上遊的表結構發生了變更也會即時同步到Paimon表中。詳見Paimon資料攝入連接器章節。

Variant 讀裁剪

僅讀取查詢用到的 Variant 成員，降低 I/O 和記憶體開銷。

開啟方式

配置項	預設值	說明
`variant.read.pushdown.enabled`	false	設定為`true`啟用 Variant 讀裁剪。

前置條件

Variant 資料僅支援VVR 11.6及以上版本寫入。舊版本或其他引擎寫入的 Variant 不支援讀裁剪。
上遊若想生效，必須開啟variant.inferShreddingSchema配置項來開啟variant shredding自動推斷，或通過配置variant.shreddingSchema及其相關配置來顯式指定適合做shredding的variant成員，詳見Coreoptions。

生效情境

SQL 通過字串鍵訪問 Variant 成員時生效，例如：

SELECT v['a']            FROM t;   -- 單層
SELECT v['a']['b']       FROM t;   -- 嵌套
SELECT v['a'], v['b']    FROM t;   -- 多成員
SELECT id, v['a']        FROM t;   -- 與普通欄位混用
SELECT v['a'] + 1        FROM t;   -- 運算式中引用

不生效情境

直接引用整個 Variant，如SELECT v FROM t。
同時直接引用與成員訪問，如SELECT v, v['a'] FROM t。
數組下標訪問，如v[0]。
表 Schema 中包含嵌套 Row 欄位。此時該表內所有 Variant 欄位均不生效。

Blob預簽名URL

Paimon支援為儲存在OSS上的Blob列資料產生短時效的公網HTTPS GET預簽名URL。外部服務（例如多模態模型服務）可以通過該URL直接拉取檔案內容，無需在Flink作業內傳輸檔案位元組。

使用限制

僅Realtime Compute引擎VVR 11.9-preview1及以上版本支援。
僅支援儲存在OSS上的Paimon表，與Metastore類型無關（Filesystem、DLF等均可）。
產生的URL為公網HTTPS URL，需要由能夠訪問公網的外部服務消費。若Catalog使用內網Endpoint，建議改為公網HTTPS Endpoint，以確保外部服務可以訪問產生的URL。
讀取普通Blob列時，需先通過blob-as-descriptor選項擷取BlobDescriptor位元組，再傳入函數。

文法

descriptor_to_presigned_url(source_table, descriptor, validity)
try_descriptor_to_presigned_url(source_table, descriptor, validity)

入參

參數	資料類型	說明
`source_table`	STRING字面量	Blob列所屬的Paimon表，格式為`database.table`；當Catalog與函數所屬Catalog一致時，也可使用`catalog.database.table`。必須為非空字面量，不支援表欄位、動態列或CASE運算式。
`descriptor`	BYTES	BlobDescriptor位元組，即開啟`blob-as-descriptor` = 'true'後讀取Blob列的結果。
`validity`	INTERVAL	URL有效期間，必須為正整數秒，例如`INTERVAL '5' MINUTE`。

出參

資料類型	說明
STRING	短時效的公網HTTPS GET預簽名URL。

樣本

-- 通過SQL Hint在讀取時以descriptor形式讀取Blob列，並產生預簽名URL
SELECT sys.descriptor_to_presigned_url(
    'default.image_table',
    image,
    INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;

-- 容錯模式：行級錯誤返回NULL，其餘行為一致
SELECT sys.try_descriptor_to_presigned_url(
    'default.image_table',
    image,
    INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;

說明

descriptor_to_presigned_url在行級錯誤時拋出異常；try_descriptor_to_presigned_url在行級錯誤時返回NULL。
產生的URL不帶副檔名，URL對應的位元組內容與Blob資料一致。下遊服務應依據返回的位元組內容識別檔案格式，不應依賴URL尾碼。
預簽名URL等價於臨時訪問憑證：產生後應立即提交給下遊服務使用，不要寫入日誌、結果表，也不要持久化儲存。

鏈式表

鏈式表（Chain Table）把主鍵表的資料拆成兩部分：全量資料寫入snapshot分支，兩次全量之間的變化資料寫入delta分支。讀取時以最近的快照分區為錨點，合并其後的增量分區重建全量資料，無需每次重寫整表。典型情境為ODS binlog全量備份後按天追加增量分區。詳情請參見Apache Paimon鏈式表。僅Realtime Compute引擎VVR 11.9及以上版本支援鏈式表。

開啟鏈式表

建表時設定chain-table.enabled為true，建立snapshot與delta分支，再為主表和兩個分支表配置分支回退參數。

CREATE TABLE `default`.`t` (
  `t1` STRING,
  `t2` STRING,
  `t3` STRING,
  `date` STRING,
  PRIMARY KEY (`date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`date`) WITH (
  'chain-table.enabled' = 'true',
  'sequence.field' = 't2',
  'bucket-key' = 't1',
  'bucket' = '2',
  'partition.timestamp-pattern' = '$date',
  'partition.timestamp-formatter' = 'yyyyMMdd'
);

-- mycat為Paimon Catalog名稱
CALL `mycat`.sys.create_branch('default.t', 'snapshot');
CALL `mycat`.sys.create_branch('default.t', 'delta');

ALTER TABLE `default`.`t` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

ALTER TABLE `default`.`t$branch_snapshot` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

ALTER TABLE `default`.`t$branch_delta` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

寫入資料

全量資料寫入t$branch_snapshot，每個分區儲存該時間點視圖。增量資料寫入t$branch_delta，每個分區僅儲存該周期的變化資料。

-- 全量寫入
INSERT OVERWRITE `default`.`t$branch_snapshot` PARTITION (date = '20250810') VALUES ('1', '1', '1');

-- 增量寫入
INSERT INTO `default`.`t$branch_delta` PARTITION (date = '20250811') VALUES ('2', '1', '1');

讀取資料

讀取方式	操作	結果
全量查詢	查詢主表指定分區。	分區的全量資料。快照分支已有該分區時直接讀取，否則由錨點快照與其後增量合并重建。
增量查詢	查詢`t$branch_delta`指定分區。	僅該分區的增量資料，不重建全量狀態。
混合查詢	全量查詢與增量查詢`UNION ALL`。	全量與增量結果的並集，同一條資料可能重複出現。
流式讀取	流模式讀取主表。	先輸出全量載入階段結果，再持續消費增量分支的新增提交。
維表查詢	作為維表進行Lookup Join。	反映各分組最新快照分區與其後增量分區的合并狀態，並按`continuous.discovery-interval`周期性增量重新整理。

-- 全量查詢
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811';

-- 增量查詢
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';

-- 混合查詢
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811'
UNION ALL
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';

流式讀取

流式讀取分兩個階段：

全量載入階段：預設讀取每個分組最新的快照分區與其後的增量分區，較早的快照分區不參與。需要在此階段按錨點合并快照與增量、消化跨分支的DELETE記錄時，將chain-table.streaming.merge-snapshot設定為true，代價是啟動階段掃描開銷增大。
增量階段：持續讀取delta分支的新增提交。

SET 'execution.runtime-mode' = 'streaming';

INSERT INTO downstream_sink SELECT * FROM `default`.`t`;

說明

增量階段僅監聽delta分支，快照分支的新寫入需重啟作業後才會被感知。建議定期通過compact_chain_table預存程序將指定分區的增量資料合併進快照分支，合并後全量載入階段僅需處理合并之後新到達的增量資料。

多分區維度

表存在多個分區維度時，可用chain-table.chain-partition-keys劃分鏈式維度與分組維度。取值必須是分區鍵的連續尾碼：尾碼欄位構成按時間排序的鏈，其前的分區欄位作為分組維度，各分組獨立維護自身的鏈式關係與錨點。未配置時，全部分區歸屬同一隱式分組。

CREATE TABLE `default`.`t` (
  `t1` STRING,
  `t2` STRING,
  `t3` STRING,
  `region` STRING,
  `date` STRING,
  PRIMARY KEY (`region`, `date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`region`, `date`) WITH (
  'chain-table.enabled' = 'true',
  'sequence.field' = 't2',
  'bucket-key' = 't1',
  'bucket' = '2',
  'partition.timestamp-pattern' = '$date',
  'partition.timestamp-formatter' = 'yyyyMMdd',
  -- 僅date參與鏈式合并，region各成一組
  'chain-table.chain-partition-keys' = 'date'
);

使用限制

僅支援主鍵表，需指定bucket與bucket-key。
各分支的表結構必須保持一致。
需指定sequence.field，不支援first-row與aggregation合并機制。
流式讀取與維表查詢要求delta分支使用deduplicate（預設）合并機制，批量讀取不受該限制。
刪除向量僅支援deduplicate合并機制的鏈式表。
changelog-producer僅支援none（預設）與input，不支援lookup與full-compaction。取值為none時按含鏈式分區在內的完整主鍵歸一化資料，原記錄與回撤記錄落在不同分區的-D、-U記錄會被丟棄；下遊需要接收跨分區回撤記錄時，請設定為input。
流式讀取與維表查詢僅支援預設啟動模式（latest-full）。指定scan.snapshot-id、scan.timestamp-millis、scan.mode為latest或consumer-id等起始位點時報錯。需要不含鏈式邏輯的標準流讀時，請直接讀取分支表（例如t$branch_delta）。
流式讀取與維表查詢不支援分區過濾，包括在WHERE中指定分區列和使用scan.partitions參數。分區過濾會干擾跨分支的鏈式合并邏輯，讀取指定分區請使用批模式。
維表查詢的Join鍵不能包含分區鍵，否則報錯。
partition.timestamp-formatter須指定完整日期（年月日或年積日），僅使用可往返解析的日期時間欄位（y/u、M/L、d、D、H、k、m、s），且最小步長不小於1秒。

Mosaic檔案格式

Mosaic是面向超寬表（萬級列）的列式分桶混合格式。列按名稱排序後以範圍方式分入若干列桶，桶內列式儲存並按桶獨立ZSTD壓縮。列裁剪時只解壓包含目標列的列桶，大幅降低寬表讀取的讀放大；名稱首碼相近的列落入同一列桶，兼顧壓縮率與列裁剪局部性。詳情請參見Mosaic。僅Realtime Compute引擎VVR 11.9及以上版本支援Mosaic格式。

開啟方式

建表時將file.format設定為mosaic。資料攝入作業對應table.properties.file.format。

CREATE TABLE `default`.`wide_table` (
  `id` BIGINT,
  `col_a` STRING,
  `col_b` DOUBLE
) WITH (
  'file.format' = 'mosaic'
);

格式參數

配置項	預設值	說明
`mosaic.num-buckets`	auto	列桶數量，用於並行I/O。取值為0或不指定時由格式自動決定列桶數量。
`mosaic.stats-columns`	（空）	需要採集min/max統計資訊以支援謂詞下推的列名，多個列用英文逗號（,）分隔。為空白時不採集統計資訊。
`mosaic.read.prefetch-row-groups`	8	讀端提前開啟的Row Group數量，用於將Range讀的延遲與解碼過程重疊。每個提前開啟的Row Group會在記憶體中保留解碼後的資料並佔用獨立輸入資料流。取值為0時關閉預取。
`mosaic.read.prefetch-max-bytes`	64 mb	預取Row Group解碼後估算大小的上限。列裁剪範圍大或Row Group大時，實際生效的`mosaic.read.prefetch-row-groups`會低於配置值。

使用限制

Mosaic不支援ARRAY、MAP、MULTISET、ROW、VARIANT、BLOB、VECTOR等複雜類型。

使用 Paimon 維表

Paimon 表支援作為維表使用。關於維表 JOIN 的文法，請參見維表JOIN語句。

Lookup 預設在每個並發載入全量資料，僅適用於小資料量的維表。巨量資料量情境請參見下文的 Shuffle Lookup 方案。

分區維表

如果維表是分區表，且僅需使用最新的一個或兩個分區資料，可使用維表分區動態載入功能：

SELECT * FROM T
JOIN DIM /*+ OPTIONS('lookup.dynamic-partition'='max_pt()', 'lookup.dynamic-partition.refresh-interval'='1 h') */
FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col = D.col;

參數	資料類型	預設值	備忘
lookup.dynamic-partition	String	無	`max_pt()`：僅載入最新分區。`max_two_pt()`：僅載入最新兩個分區。
lookup.dynamic-partition.refresh-interval	Duration	1 h	檢查維表分區更新的時間間隔。

巨量資料量維表：固定分桶表

僅VVR 8.0.8及以上版本支援。對於固定分桶表（bucket > 0），可通過 SHUFFLE Lookup 將資料按分桶鍵分散到各並發，每個並發僅載入對應桶的資料：

SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

說明

串連欄位必須是 bucket key。bucket key 預設值為主鍵。
僅固定分桶表（bucket > 0）支援此功能。

巨量資料量維表：非固定分桶表

僅VVR 8.0.10及以上版本支援。對於動態分桶表或 Append 表，可通過 SHUFFLE_HASH 或 REPLICATED_SHUFFLE_HASH 實現每個並發僅保留所需的部分資料：

-- Shuffle Hash
SELECT /*+ SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

-- Replicated Shuffle Hash
SELECT /*+ REPLICATED_SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

關於 SHUFFLE_HASH 和 REPLICATED_SHUFFLE_HASH 的詳細說明，請參見維表JOIN語句。

資料攝入

Paimon連接器可以用於資料攝入YAML作業開發，作為目標端寫入。

文法結構

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

配置項

參數	說明	是否必填	資料類型	預設值	備忘
type	連接器類型。	是	STRING	無	固定值為`paimon`。
name	目標端名稱。	否	STRING	無	Sink的名稱。
catalog.properties.metastore	Paimon Catalog的類型。	否	STRING	filesystem	取值如下： filesystem（預設值） rest（僅支援DLF，不支援DLF-Legacy）
catalog.properties.*	建立Paimon Catalog的參數。	否	STRING	無	詳情請參見管理Paimon Catalog。
table.properties.*	建立Paimon table的參數。	否	STRING	無	詳情請參見Paimon table options。
table.properties.file.format	表中資料檔案的儲存類型。	String	否	parquet	可選值： orc parquet avro lance (VVR 11.6及以上版本） mosaic (VVR 11.9及以上版本)
catalog.properties.warehouse	檔案儲存體的根目錄。	否	STRING	無	僅在`catalog.properties.metastore`設定為 filesystem時生效。
commit.user-prefix	提交資料檔案時的使用者名稱首碼。	否	STRING	無	說明建議為不同的作業設定不同的使用者名稱，方便在出現提交衝突時定位衝突的作業。
partition.key	每個分區表的分區欄位。	否	STRING	無	不同的表使用`;`分割，不同的欄位使用`,`分割，表與欄位使用`:`分割。例如：`testdb.table1:id1,id2;testdb.table2:name`。
sink.cross-partition-upsert.tables	指定需要跨分區更新（主鍵不包含所有分區欄位）的表。	否	STRING	無	適用於跨分區更新的表。格式：使用分號`;`分隔表名。效能建議：此操作消耗大量計算資源。請為這些表建立獨立的作業。重要必須列出所有合格表。遺漏表名將導致資料重複。
sink.commit.parallelism	指定Commit運算元的並發度。	否	INTEGER	無	在Commit運算元成為瓶頸時可以通過該參數指定Commit運算元的並發度以提升效能。僅Realtime Compute引擎VVR 11.6及以上版本支援該參數。說明設定該參數會導致運算元並發度修改，有狀態重啟作業時需要指定允許忽略部分運算元狀態（AllowNonRestoredState）。
sink.existing-table.schema-check.enabled	寫入預先建立的表時，是否對已有表結構進行相容性校正與自動適配。	Boolean	否	true	取值及寫入預先建立的表時的行為如下： true（預設值）：校正已有表的列類型相容性與 NOT NULL 列缺失/多餘，並自動補齊上遊存在而下遊缺失的可空列；出現無法自動適配的情況時作業報錯並給出建議執行的 SQL。 false：跳過上述校正與自動補列，直接以下遊已有表結構為準寫出。僅跳過執行初始建表語句時的補列操作，後續的表結構變更會自動執行。說明僅Realtime Compute引擎VVR 11.8及以上版本支援。

複用已有 Catalog

自VVR 11.5版本起，您可以在Flink CDC資料攝入作業中直接引用“資料管理”頁面中建立的內建Paimon Catalog，減少手寫串連屬性工作量。

sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com

目前，資料攝入作業支援自動複用所有 Paimon Catalog 參數，等價於在 YAML 作業中手動設定 catalog.properties.首碼的參數。

如果希望覆蓋以上自動複用的參數，可顯式寫出相應的 YAML 參數，其具備更高的優先順序。例如，上面的例子中 Catalog 參數 fs.oss.endpoint總是採納 YAML 作業中顯式寫出的選項，而非在 paimon_dlf_catalog中定義的選項。

表結構變更

CDC Yaml Pipeline作業在處理表結構變更時有不同的策略，通過pipeline層級的配置項schema.change.behavior來進行表結構變更同步配置。schema.change.behavior取值有IGNORE、LENIENT、TRY_EVOLVE、EVOLVE 和 EXCEPTION。其中LENIENT和EVOLVE涉及到表結構變更，接下來會說明如何處理不同表結構變更事件（Schema Change Event）。

LENIENT（預設）

LENIENT模式下支援的Schema變更策略詳情如下：

添加可空列：會自動在結果表Schema末尾添加對應的列，並自動同步新增列的資料。
刪除可空列：不會直接在結果表中刪除該列，而是將該列的資料自動填滿為NULL值。
添加非空列：會自動在結果表Schema末尾添加對應的列，並自動同步新增列的資料，新增的列會預設設定為可空列，對於添加列發生之前的資料自動化佈建為NULL值。
重新命名列：被看作為添加列和刪除列。直接在結果表中末尾添加重新命名後的列，並將重新命名前的列資料自動填滿為NULL值。例如，如果col_a重新命名為col_b，則會在結果表末尾添加col_b，並自動將col_a的資料填充為NULL值。
列類型變更：會在結果表中變更對應列的資料類型。
暫不支援同步以下Schema的變更：
- 主鍵類型的變更。
- 從NULLABLE轉為NOT NULL變更。

EVOLVE

EVOLVE模式下支援的Schema變更策略詳情如下：

添加可空列：支援
刪除可空列：不支援。
添加非空列：會在結果表添加可空列。
重新命名列：支援，會在結果表將原有列重新命名。
列類型變更：會在結果表中變更對應列的資料類型。
暫不支援同步以下Schema的變更：
- 主鍵類型的變更。
- 從NULLABLE轉為NOT NULL變更。

開啟EVOLVE模式樣本請參見開啟EVOLVE模式。

說明

在下遊 Paimon 表已經存在時，會優先使用已有的表結構進行寫入，不會嘗試重複建表。在已有的表結構和上遊表不一致時：上遊存在而下遊缺失的可空列會被自動添加到已有表；若出現列類型不相容、NOT NULL 列缺失或多餘等無法自動適配的情況，會出現校正不通過的報錯，您可以參考報錯中建議執行的 SQL 修改表結構，也可以通過在 sink 中配置 sink.existing-table.schema-check.enabled 為 false 忽略該校正與自動補列，保持以下遊表結構為目標進行寫出。

使用樣本

下面展示幾個典型的使用者使用情境下的配置樣本。

寫入 Rest Catalog

Paimon Catalog為rest，寫入阿里雲資料湖構建的配置樣本：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true

其中，catalog.properties首碼的參數含義請參見Flink CDC對接Catalog配置參數。

寫入 FileSystem Catalog

Paimon Catalog為filesystem，寫入阿里雲OSS的配置樣本：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true

其中，catalog.properties首碼的參數含義請參見建立Paimon Filesystem Catalog。

寫入分區表

將上遊時間戳記類型的create_time欄位轉化為日期類型，作為Paimon表的分區鍵。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true
 
transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Paimon Pipeline

單表同步

建議為對即時性和穩定性有較高要求的表配置單表同步作業，能夠避免多表同步帶來的複雜性風險，單表同步的配置樣本如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

整庫同步

使用整庫同步能力能夠降低需要配置和啟動並執行同步作業數，減少營運成本，配置樣本如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

整庫同步並替換庫表名

在整庫同步的基礎上，可以增加 route 模組對庫名、表名進行統一替換，配置樣本如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true
  
route:
  # 將 MySQL test_db 資料庫中的所有表同步到 Paimon test_db2 庫中，並保留表名不變
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Paimon Pipeline

分庫分表合并

在您有將多張分表的資料寫入到一張下遊表需求的情境下，可以使用如下的配置模板：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true
  
route:
  # MySQL test_db 中的所有分表將被合并到一張 Paimon test_db.user 表中。
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Paimon Pipeline

重啟新增存量表

如果想要新增同步存量表，設定scan.newly-added-table.enabled = true後重啟作業。

重要

如果作業先設定scan.binlog.newly-added-table.enabled為true捕獲新增表，不可以再通過scan.newly-added-table.enabled = true重啟捕獲存量表，否則會有資料重複發送。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # 在作業重啟時，檢查 tables 參數捕獲的新表並執行快照。
  # 注意需要搭配 scan.startup.mode: initial 參數使用。
  scan.newly-added-table.enabled: true
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

整庫同步時排除部分表

在整庫同步的情境下，如果有部分表（測試表或者某些會導致作業持續 Failover 的表）需要排除，配置樣本如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # 此正則命中的表都不會被同步。
  tables.exclude: test_db.table1
  server-id: 5401-5499
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

整庫同步到檔案格式為Lance的Paimon表

Lance是一種專為向量和多模態資料設計的資料存放區格式，您可以通過設定file.format的建表參數以寫到檔案格式為Lance的Paimon表，配置樣本如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true
  # 指定寫出檔案格式為 lance 
  table.properties.file.format: lance
pipeline:
  name: MySQL to Paimon Pipeline

開啟EVOLVE模式

開啟 EVOLVE 模式能夠保持下遊表與上遊表表結構嚴格一致，包括執行與上遊相同的刪除欄位、刪除表等操作。在此模式下，若下遊無法支援應用所有的表結構變更事件，可能導致作業Failover且無法自愈。作業配置樣本如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可選）提交使用者名稱，建議為不同作業設定不同的提交使用者以避免衝突
  commit.user: your_job_name
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline
  schema.change.behavior: evolve

Paimon社區版本	Realtime ComputeFlink版引擎版本（VVR ）
1.3.1	11.5、11.6、11.7、11.8
1.3	11.4
1.2	11.2、11.3
1.1	11.1
1.0	8.0.11

背景資訊

特色功能

使用限制與建議

SQL

文法結構

WITH參數

向量表專屬參數

特色功能詳解

資料新鮮度與一致性保證

資料合併機制

增量資料產生機制

寫入模式

作為Flink CDC資料攝入的目標端

Variant 讀裁剪

Blob預簽名URL

鏈式表

Mosaic檔案格式

使用 Paimon 維表

分區維表

巨量資料量維表：固定分桶表

巨量資料量維表：非固定分桶表

資料攝入

文法結構

配置項

複用已有 Catalog

表結構變更

LENIENT（預設）

EVOLVE

使用樣本

寫入 Rest Catalog

寫入 FileSystem Catalog

寫入分區表

單表同步

整庫同步

整庫同步並替換庫表名

分庫分表合并

重啟新增存量表

整庫同步時排除部分表

整庫同步到檔案格式為Lance的Paimon表

開啟EVOLVE模式

常見問題