Connector 參數 - Realtime Compute for Apache Flink

本文將為您介紹Fluss Connector的相關參數。

如何配置參數

建立表時配置參數

一旦表被建立後，這部分定義的參數會被作業表的中繼資料持久化儲存在 Fluss 中。其中儲存參數在讀寫資料的時候會生效。

# 設定table.log.ttl為7天並且在讀取時禁止CRC校正
CREATE TABLE `my-catalog`.`my_db`.`my_log_table` (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) WITH (
  'table.log.ttl' = '7d',
  'client.scanner.log.check-crc' = 'false'
);

重要

已持久化到 Fluss 中繼資料中的表（即通過 DDL 語句 CREATE TABLE 建立的登入表），不支援直接修改登入的 Connector 參數（如 client.writer.batch-size、client.writer.buffer.memory-size 等）。即使在控制台中點擊編輯按鈕，上述參數修改也無法生效。

SQL Hints 動態配置參數

Fluss 也支援通過 SQL Hints 的方式動態配置參數。這部分動態配置參數不會作為中繼資料的一部分持久化儲存在 Fluss 中。這部分配置只會在當前 query 中生效而不會影響其他的 query。

重要

動態配置的參數的優先順序要高於表中繼資料中持久化的參數。
儲存參數不支援通過 SQL Hints 動態配置。

-- 讀取時設定 SQL hints
SELECT * FROM `my-catalog`.`my_db`.`my_log_table` /*+ OPTIONS('client.scanner.log.check-crc' = 'false') */;

-- 寫入時設定 SQL hints
INSERT INTO `my-catalog`.`my_db`.`my_pk_table_2` /*+ OPTIONS('sink.ignore-delete'='true') */ select * from `my-catalog`.`my_db`.`my_pk_table_1`;

儲存參數

參數	類型	預設值	說明
bucket.num	int	1	Fluss叢集的桶數量。
bucket.key	String	無	資料將根據`bucket.key`的雜湊值分布到各個桶中。`bucket.key`必須是主鍵的一個子集（不包含分區鍵）。若指定多個欄位作為 bucket.key，欄位之間使用逗號 `,` 分隔。若表定義了主鍵且未顯式指定`bucket.key`，則預設使用主鍵（排除分區鍵）作為`bucket.key`。若表未定義主鍵且也未指定`bucket.key`，則資料將在各桶間隨機分布。
table.log.ttl	Duration	7d	日誌資料的存留時間。該配置控制日誌保留的最大時間，之後將刪除舊的日誌資料以釋放空間。如果設定為 -1，則日誌不會被刪除。
table.auto-partition.enabled	Boolean	false	啟用自動分區後，表的分區將自動建立。
table.auto-partition.time-unit	ENUM	DAY	自動建立分區的時間粒紋。有效值為`YEAR`，`QUARTER`，`MONTH`，`DAY`，`HOUR`。例如：值為`HOUR`，自動建立分區的格式為`yyyyMMddHH`。值為`DAY`，自動建立分區的格式為`yyyyMMdd`。
table.auto-partition.num-precreate	Integer	2	每次檢查自動分區時預建立的分區數量。例如，如果當前檢查時間為 2024-11-11，且值配置為 3，則將預建立分區 20241111、20241112 和 20241113。如果任何一個分區已存在，則跳過建立該分區。預設值為 2，表示將預建立 2 個分區。如果 table.auto-partition.time-unit 為 DAY（預設值），一個預建立的分區為當天，另一個為明天。
table.auto-partition.num-retention	Integer	7	每次檢查自動分區時保留的歷史分區數量。例如，如果當前檢查時間為 2024-11-11，時間單位為 DAY，且值配置為 3，則將保留歷史分區 20241108、20241109 和 20241110。早於 20241108 的分區將被刪除。預設值為 7，表示將保留 7 個分區。
table.auto-partition.time-zone	String	系統時區	自動分區的時區，預設與系統時區相同。
table.replication.factor	Integer	（無）	日誌表的副本數量。如果不設定，Fluss 將使用叢集預設的副本數量 default.replication.factor。該值必須是正數，且不大於 Fluss 叢集中 tabletServer 的數量。如果設定的值大於 Fluss 叢集中的 tabletServer 數量，則在建立新表時會報錯。
table.log.format	Enum	ARROW	日誌儲存中日誌儲存的格式。預設值為 ARROW。支援的格式為 ARROW 和 INDEXED。
table.log.arrow.compression.type	Enum	ZSTD	如果日誌格式設定為 ARROW，則該值時配置日誌資料的壓縮類型。候選的壓縮類型為 NONE、LZ4_FRAME 和 ZSTD。預設值為 ZSTD。
table.log.arrow.compression.zstd.level	Integer	3	如果日誌格式設定為 ARROW 且壓縮類型設定為 ZSTD，則該值配置壓縮的層級。有效範圍為 1 到 22。預設值為 3。
table.statistics.columns	String	（無）	指定需要採集列統計資訊的列名列表，多個列以逗號分隔（如 `'temperature,location'`）。設定為 `*` 表示採集所有支援類型的列。列統計資訊用於謂詞下推最佳化，僅適用於 Arrow 日誌格式的日誌表。支援的資料類型包括：BOOLEAN、TINYINT、SMALLINT、INT、BIGINT、FLOAT、DOUBLE、DECIMAL、STRING、CHAR、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ。不支援的類型（BYTES、BINARY、ARRAY、MAP、ROW）會被自動排除。詳情請參見謂詞下推。
table.kv.format	Enum	COMPACTED	kv 儲存中 kv 資料的儲存格式。預設值為 COMPACTED。支援的格式為 COMPACTED 和 INDEXED。
table.log.tiered.local-segments	Integer	2	啟用日誌階層式存放區時，每個表在本地保留的日誌段（Log Segment）數量。必須大於 0，預設值為 2。
table.datalake.enabled	Boolean	false	是否啟用表的湖儲存，預設禁用。啟用後，若湖倉分層服務已啟動，則表資料將自動分層並壓縮至湖儲存中。
table.datalake.format	Enum	（無）	指定表所使用的資料湖格式，用於分層湖倉儲存。當前僅支援`paimon`。該參數的核心作用是：確保 Fluss 與外部湖倉系統在鍵編碼（key encoding）和分桶策略（bucketing）上保持一致；支援 Union Read 功能，實現 Fluss 表與資料湖表的統一讀取。配置行為說明優先順序最高：若在建表時顯式設定 table.datalake.format，則以該值為準。預設值：若未設定，則使用 Fluss 叢集級配置的 datalake.format 值。動態啟用支援：可在 table.datalake.enabled = false 時預先配置 table.datalake.format，後續通過啟用 table.datalake.enabled 來啟用資料湖功能，無需重建表。
table.merge-engine	Enum	（無）	定義主鍵表的資料更新策略。預設策略為保留最新資料（last_row）。此外，支援以下兩種策略： first_row：保留相同主鍵的第一條資料。 versioned：保留相同主鍵中版本號碼最大的資料。
table.merge-engine.versioned.ver-column	String	（無）	如果合并原則設定為 versioned，則必須設定版本列的列名。
table.log.local-ttl	Duration	無	本地日誌分段的保留時間長度。超過該時間長度後，本地日誌分段方可被刪除。開啟遠程日誌分層時，已到期的本地分段需先複製到遠程儲存才會刪除。設定為 `0ms` 時關閉基於 TTL 的刪除。未配置時繼承 `table.log.ttl`；兩者均為正值時，該值必須小於或等於 `table.log.ttl`。
table.kv.format-version	Integer	無	KV 格式版本。未配置時由 Coordinator 在建表時自動化佈建。下述資料湖編碼與分桶策略僅在叢集級配置了 `datalake.format` 時生效。版本 1：該參數引入前建立的表按版本 1 處理，主鍵與 Bucket Key 均使用資料湖編碼器（如 Paimon、Iceberg）。部分資料湖編碼器不保證 Bucket Key 編碼位元組是主鍵編碼位元組的首碼，因此可能無法正確支援首碼查詢。版本 2（當前）：Bucket Key 與主鍵不同時，建立表的主鍵編碼使用 Fluss 預設編碼器，可正確支援首碼查詢；Bucket Key 與主鍵相同時，仍使用資料湖編碼器以最佳化湖側寫入。Bucket Key 編碼始終使用資料湖編碼器，以對齊資料湖的分桶計算。
table.changelog.image	Enum	FULL	主鍵表的 changelog 鏡像模式，語義參照 MySQL `binlog_row_image` 與 PostgreSQL `replica identity`。僅適用於主鍵表。 `FULL`（預設）：更新操作同時產生 UPDATE_BEFORE 與 UPDATE_AFTER 記錄，保留完整變更資訊，可追溯更新前的值。 `WAL`：不產生 UPDATE_BEFORE 記錄，僅輸出 INSERT、UPDATE_AFTER 以及允許時的 DELETE。使用預設 Merge Engine、整行更新（非 Partial Update）且無自增列時，跳過舊值尋找，INSERT 操作轉換為 UPDATE_AFTER 事件。該模式降低儲存與傳輸開銷，但無法追溯更新前的值。
table.delete.behavior	Enum	ALLOW	主鍵表刪除操作的行為。為下遊鏈路（如 Flink Delta Join）提供系統級保證，確保表 changelog 中不出現刪除事件。僅適用於主鍵表。 `ALLOW`（預設 Merge Engine 下的預設值）：允許正常刪除。 `IGNORE`：靜默忽略刪除請求，不報錯。 `DISABLE`：拒絕刪除請求並拋出明確錯誤。 Merge Engine 為 `first_row`、`versioned` 或 `aggregation` 的表，該參數自動設為 `IGNORE` 且不可覆蓋。`aggregation` Merge Engine 下設為 `ALLOW` 時，刪除操作會移除整條記錄。
table.auto-inc.batch-size	Long	100000	每次從分布式計數器擷取自增 ID 的批量大小，決定本機快取 ID 區間的長度。取值過大時，TabletServer 重啟或異常退出會丟棄未使用的緩衝區間，造成明顯的自增 ID 空洞。取值過小時，向分布式計數器請求 ID 的頻率上升，引入額外網路開銷，寫入吞吐下降。
table.datalake.auto-compaction	Boolean	false	是否在分層服務寫入資料湖時自動觸發 compaction。
table.datalake.auto-expire-snapshot	Boolean	false	是否在分層服務向資料湖提交時自動觸發快照到期。
table.datalake.historical-partition.enabled	Boolean	false	是否為該表啟用歷史分區查詢。開啟後，Coordinator 建立並保留一個系統磁碟分割，用於把已到期分區的維表查詢路由至湖儲存。當前僅支援單分區鍵的自動分區 Paimon 主鍵表。修改該參數後，需重啟需要查詢歷史分區資料的維表作業，使用戶端載入新的表配置。

讀取參數

參數	類型	預設值	說明
scan.startup.mode	Enum	full	讀模數式允許您指定資料消費的起始點。Fluss 目前支援以下 scan.startup.mode 選項：full（預設值）、earliest、latest、timestamp。更多詳情請參見讀取資料。
scan.startup.timestamp	Long	（無）	開始讀取資料的時間戳記。僅當 `scan.startup.mode` 設定為 `timestamp` 時，此選項才有效。格式為 `milli-second-since-epoch` 或 `yyyy-MM-dd HH:mm:ss`，例如`1678883047356`或 `2023-12-09 23:09:12`。說明該參數不支援使用 Flink 內建時間函數（如 CURRENT_TIMESTAMP）動態產生時間值。WITH 子句中的連接器參數僅接受固定字面值，無法解析 SQL 函數運算式。如需傳入動態時間，可在作業外部通過變數注入方式傳入固定時間戳記或毫秒數。
scan.partition.discovery.interval	Duration	10s	在讀取分區表時，Fluss Source 可按照設定的時間間隔自動探測新分區。當檢測到新分區時，會動態將其加入讀取計劃；若分區被刪除，則自動取消訂閱。若將該間隔設為負值，則表示禁用自動分區發現功能。
client.scanner.log.check-crc	Boolean	true	是否對讀取到的訊息執行 CRC32 校正。該功能用於確保訊息在傳輸或磁碟預存程序中未發生損壞，提升資料可靠性。雖然會引入極小的計算開銷，但鑒於其對資料完整性的關鍵作用，建議在生產環境中保持啟用。
client.scanner.log.max-poll-records	Integer	500	LogScanner 的單次 poll() 調用返回的最大記錄數。請注意，此配置不會影響底層的擷取行為。LogScanner 將緩衝每次擷取請求的記錄，並從每次 poll 中逐步返回它們。
client.scanner.log.fetch.max-bytes	MemorySize	16mb	用戶端從伺服器擷取請求的最巨量資料量。記錄是分批擷取的，如果第一個非空桶的第一個記錄批次大於此值，該記錄批次仍將被返回，以確保擷取可以取得進展。因此，這不是一個絕對的最大值。
client.scanner.log.fetch.max-bytes-for-bucket	MemorySize	1mb	用戶端從伺服器擷取請求時，每個表桶返回的最巨量資料量。記錄是分批擷取的，最大位元組大小由此選項配置。
client.scanner.log.fetch.min-bytes	MemorySize	1b	用戶端每次擷取日誌請求從伺服器響應的期望最小位元組數。如果位元組數不足，將等待最多 client.scanner.log.fetch-wait-max-time 時間再返回。
client.scanner.log.fetch.wait-max-time	Duration	500ms	用戶端每次擷取日誌請求從伺服器響應的最大等待時間。
client.scanner.io.tmpdir	String	System.getProperty("java.io.tmpdir") + "/fluss"	用戶端用於臨時儲存資料檔案（如 kv 快照、日誌段檔案）的本地目錄，以便讀取。
client.scanner.remote-log.prefetch-num	Integer	4	LogScanner 從遠程儲存下載的遠程日誌段數量，這些日誌段將被儲存在本地臨時檔案中。預設設定為 4。
client.remote-file.download-thread-num	Integer	3	用戶端用於下載遠程檔案的線程數。
scan.kv.snapshot.lease.id	String	UUID	用於保護已擷取 KV 快照不被刪除的租約 ID。指定後，快照保留至消費端處理完全部快照，或租約時間長度到期。未顯式指定時，取值為隨機產生的 UUID 字串。
scan.kv.snapshot.lease.duration	Duration	1day	KV 快照租約的到期等待時間長度，避免快照阻塞刪除。

寫入參數

參數	類型	預設值	說明
sink.ignore-delete	Boolean	false	如果設定為 true，則目標端將忽略 DELETE 和 UPDATE_BEFORE 的變更日誌事件。
sink.bucket-shuffle	Boolean	true	是否在寫入目標端之前按桶 ID 進行洗牌。將具有相同桶 ID 的資料洗牌到由同一任務處理可以提高用戶端處理效率並減少資源消耗。對於日誌表，只有在定義了 bucket.key 時，bucket shuffle才會生效。對於主鍵表，預設啟用。
client.writer.buffer.memory-size	MemorySize	64mb	寫入器可以用來緩衝內部行的總位元組數。
client.writer.batch-size	MemorySize	2mb	寫入器將同一桶的記錄批量組合在一起。這有助於提高用戶端的效能。將嘗試將同一桶的記錄批量組合在一起。這有助於提高用戶端和伺服器的效能。
client.writer.buffer.wait-timeout	Duration	2^(63)-1ns	定義寫入器在等待段可用時阻塞的時間。
client.writer.batch-timeout	Duration	100ms	寫入器將嘗試將到達的行批量組合在一起，以便在請求發送之間發送。通常，這僅在負載較高且行到達速度比發送速度快時才會發生。然而，在某些情況下，寫入器可能希望即使在中等負載下也減少請求數量。此設定通過添加少量人工延遲來實現這一點，即，而不是立即發送一行，寫入器將等待最多指定的延遲時間，以便其他記錄到達，以便可以將它們批量發送。這可以類比為 TCP 中的 Nagle 演算法。此設定為批量的延遲上限：一旦我們為一個桶積累了 client.writer.batch-size 的行，它將立即發送，無論此設定如何，但如果為該桶積累的位元組數少於此值，我們將等待指定的時間，等待更多記錄到達。
client.writer.bucket.no-key-assigner	Enum	STICKY	指定無鍵表的桶分配策略。對於有桶鍵或主鍵的表，系統會基於鍵的雜湊值選擇桶。對於無鍵表，可通過此參數選擇桶分配器。支援以下兩種策略： STICKY：僅當記錄批次發生變化時分配新桶 ID，否則保持與前一條記錄相同的桶 ID。 ROUND_ROBIN：按輪詢方式為輸入行分配桶 ID。
client.writer.acks	String	all	取值如下： -1（all）：Leader必須等待所有同步複本（ISR）確認後才返回成功。只要至少一個副本存活，資料就不會丟失。 1：Leader將記錄寫入本地日誌後立即確認，無需等待副本同步，若Leader在同步前故障，資料可能丟失。 0：寫入器不等待伺服器的任何確認。無法保證記錄已被接收，網路錯誤或伺服器故障可能導致資料丟失。適合高吞吐、可容忍丟資料的情境。說明生產環境建議使用all，以確保資料高可用和不丟失。
client.writer.request-max-size	MemorySize	10mb	請求的最大大小（以位元組為單位）。此設定將限制寫入器在單個請求中發送的記錄批次數量，以避免發送過大的請求。請注意，此重試與寫入器在收到錯誤後重新發送行沒有區別。
client.writer.retries	Integer	Integer.MAX_VALUE	設定大於零的值，用戶端會重發任何因可能的可重試的錯誤而發送失敗的 record。
client.writer.enable-idempotence	Boolean	true	是否啟用等冪性保障，預設為開啟。啟用後，可確保用戶端寫入的資料具有順序性和 exactly once 保證。若關閉此功能，當因 Fluss 服務端故障等原因觸發用戶端重試時，可能導致相同訊息被重複寫入流中。
client.writer.max-inflight-requests-per-bucket	Integer	5	每個桶的未確認請求的最大數量。此配置只有在 client.writer.enable-idempotence 設定為 true 時才有效。當每個桶的未確認請求數量超過此設定時，寫入器將等待未確認請求完成，然後發送新請求。
client.writer.buffer.page-size	MemorySize	128kb	寫入緩衝區（`client.writer.buffer.memory-size`）中每個記憶體頁的大小。
client.writer.buffer.per-request-memory-size	MemorySize	16mb	寫入端單次分配的最小位元組數，向下取整至 `client.writer.buffer.page-size` 的最近整數倍，且必須大於或等於 `client.writer.buffer.page-size`。按批分配記憶體可保證記憶體段連續，提升 CPU 快取命中率。
client.writer.dynamic-batch-size.enabled	Boolean	true	寫入端是否按實際寫入吞吐動態調整批大小。預設開啟。開啟後，寫入端依據目標表或分區的歷史寫入量調整每批的記憶體配置，在吞吐波動情境下提升記憶體利用率與寫入效能。動態批大小有上下界：不超過 `client.writer.batch-size`，不低於 `client.writer.buffer.page-size`。關閉後，所有批次使用固定批大小；各分區資料速率不均時，可能出現頻繁記憶體等待，寫入效能下降。
sink.distribution-mode	Enum	AUTO	資料 shuffle 至 Sink 的分發模式。 `AUTO`（預設）：按表類型自動選擇。主鍵表與定義了 Bucket Key 的日誌表使用 BUCKET 模式以最大化吞吐；無 Bucket Key 的日誌表使用 NONE 模式。 `NONE`：使用 Flink 預設 shuffle 策略。Sink 並行度與上遊相同時通常為 FORWARD，不同時為 ROUND_ROBIN。 `BUCKET`：寫入前按 Bucket ID shuffle，相同 Bucket ID 的資料由同一 Task 處理，提升用戶端處理效率並降低資源消耗。主鍵表推薦使用該模式。日誌表僅在定義了 `bucket.key` 時生效。Sink 並行度超過 Bucket 數時，部分 Sink Task 會空閑。 `PARTITION_DYNAMIC`：按分區鍵流量特徵動態調整 shuffle 策略，監控資料分布並均衡負載。僅支援分區日誌表，不支援主鍵表。分區資料扭曲或分區數較多時使用。該模式需採集資料統計並執行額外 shuffle，存在開銷。

維表參數

參數	類型	預設值	說明
lookup.async	Boolean	true	是否使用非同步Lookup。非同步Lookup比同步Lookup具有更好的輸送量效能。
`lookup.insert-if-not-exists`	Boolean	false	是否在 Lookup 未命中時自動插入新行。啟用後，當尋找鍵在維表中不存在時，Fluss 會自動插入一條以尋找鍵為主鍵的新行並返回。僅支援主鍵 Lookup，不支援 Prefix Lookup。
lookup.cache	Enum	NONE	此lookup表的緩衝策略，包括NONE、PARTIAL。
lookup.max-retries	Integer	3	Lookup操作失敗時允許的最大重試次數。
lookup.partial-cache.expire-after-access	Duration	(None)	訪問緩衝中的條目後，該條目的到期時間。
lookup.partial-cache.expire-after-write	Duration	(None)	寫入緩衝中的條目後，該條目的到期時間。
lookup.partial-cache.cache-missing-key	Boolean	true	如果lookup key在表中沒有匹配到任何行，是否將空值儲存到緩衝中。
lookup.partial-cache.max-rows	Long	(None)	緩衝中儲存的最大行數。
client.lookup.queue-size	Integer	25600	待處理lookup操作的最大數量。
client.lookup.max-batch-size	Integer	128	將多個lookup操作合并為一個lookup請求的最大批次大小。
client.lookup.max-inflight-requests	Integer	128	lookup操作中未確認的lookup請求的最大數量。
client.lookup.batch-timeout	Duration	100ms	等待lookup批次裝滿的最大時間，如果達到此逾時時間，lookup批次將被關閉並發送出去。
client.lookup.max-retries	Integer	Integer.MAX_VALUE	維表查詢請求因可重試錯誤失敗後的重發次數。取值大於 0 時，用戶端重發失敗的查詢請求。

其他參數

參數	類型	預設值	說明
bootstrap.servers	List	（無）	指定用於串連 Fluss 叢集的初始節點列表，格式為 `host1:port1,host2:port2,...`。用戶端將通過這些節點擷取完整的叢集拓撲資訊，因此列表無需包含所有伺服器。建議配置多個節點，以提高串連的高可用性，防止單點故障導致串連失敗。
client.id	String	""	設定一個具有業務意義的用戶端識別碼（如 order-service-prod），以便在伺服器日誌中清晰識別請求來源。
client.connect-timeout	Duration	120s	Netty 用戶端連線逾時時間。
client.request-timeout	Duration	30s	請求完成的逾時時間。如果使用者將寫入確認設定為 -1，則此逾時時間是延遲寫入嘗試完成的最大時間。預設設定為 30 秒。
client.filesystem.security.token.renewal.backoff	Duration	1h	在擷取檔案系統安全性權杖失敗後，重試擷取新安全性權杖的時間間隔。
client.filesystem.security.token.renewal.time-ratio	Double	0.75	在檔案系統安全性權杖到期時間的比率，用於確定何時重新擷取訪問檔案系統的新憑據。
client.metrics.enabled	Boolean	false	啟用用戶端指標。啟用指標後，用戶端將收集指標並通過 JMX 指標報告器進行報告。