Realtime Compute for Apache Flink：Kafka資料攝入連接器

配置項

• 通用

  參數	說明	是否必填	資料類型	預設值	備忘
  type	源端或目標端類型。	是	String	無	固定值為kafka
  name	源端或目標端名稱。	否	String	無	無
  properties.bootstrap.servers	Kafka broker地址。	是	String	無	格式為host:port,host:port,host:port，以英文逗號（,）分割。
  properties.*	對Kafka用戶端的直接配置。	否	String	無	尾碼名必須是Kafka官方文檔中定義的生產者和消費者配置。 Flink會將properties.首碼移除，並將剩餘的配置傳遞給Kafka用戶端。例如可以通過'properties.allow.auto.create.topics' = 'false' 來禁用自動建立topic。
  key.format	讀取或寫入Kafka訊息key部分使用的格式。	否	String	無	對Source，僅支援json。 對Sink，取值如下： csv json 說明 僅Realtime Compute引擎VVR 11.0.0及以上版本支援該參數。
  value.format	讀取或寫入Kafka訊息value部分時使用的格式。	否	String	debezium-json	對Source，取值如下： debezium-json  canal-json json 對Sink，取值如下： debezium-json  canal-json canal-protobuf 說明 僅Realtime Compute引擎VVR 8.0.10及以上版本支援debezium-json和canal-json格式。 僅Realtime Compute引擎VVR 11.0.0及以上版本支援json格式。

• 源表

  參數	說明	是否必填	資料類型	預設值	備忘
  topic	讀取的topic名稱。	否	String	無	以英文分號 (;) 分隔多個topic名稱，例如topic-1和topic-2 說明 topic和topic-pattern兩個選項只能指定其中一個。
  topic-pattern	匹配讀取topic名稱的Regex。所有匹配該Regex的topic在作業運行時均會被讀取。	否	String	無	樣本： user_event_.*：匹配所有以 user_event_ 開頭的 topic prod\.logs\..*：匹配 prod.logs. 首碼的 topic（. 需轉義） 說明 topic和topic-pattern兩個選項只能指定其中一個。
  properties.group.id	消費組ID。	否	String	無	如果指定的group id為首次使用，則必須將properties.auto.offset.reset設定為earliest或latest以指定初次開機位點。
  scan.startup.mode	Kafka讀取資料的啟動位點。	否	String	group-offsets	取值如下： earliest-offset：從Kafka最早分區開始讀取。 latest-offset：從Kafka最新位點開始讀取。 group-offsets（預設值）：從指定的properties.group.id已提交的位點開始讀取。 timestamp：從scan.startup.timestamp-millis指定的時間戳記開始讀取。 specific-offsets：從scan.startup.specific-offsets指定的位移量開始讀取。 說明 該參數在作業無狀態啟動時生效。作業在從checkpoint重啟或狀態恢複時，會優先使用狀態中儲存的進度恢複讀取。
  scan.startup.specific-offsets	specific-offsets啟動模式下，指定每個分區的啟動位移量。	否	String	無	例如partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	timestamp啟動模式下，指定啟動位點時間戳記。	否	Long	無	單位為毫秒
  scan.topic-partition-discovery.interval	動態檢測Kafka topic和partition的時間間隔。	否	Duration	5分鐘	分區檢查間隔預設為5分鐘。需要顯式地設定分區檢查間隔為非正數才能關閉此功能。開啟動態分區發現後，Kafka Source 可以自動地發現新增的分區並自動讀取對應分區上的資料。在topic-pattern模式下，不僅讀取已有topic的新增分區資料，也會讀取符合正則匹配的新增topic的所有分區資料。
  scan.check.duplicated.group.id	是否檢查通過properties.group.id指定的消費者組有重複。	否	Boolean	false	參數取值如下： true：在啟動作業前檢查消費者組是否有重複，如有重複作業將會報錯，避免與現有的消費者組產生衝突。 false：直接啟動作業，不檢查消費者組衝突。
  schema.inference.strategy	Schema解析策略。	否	String	continuous	取值如下： continuous：對每條資料均進行 Schema 解析。在前後 Schema 不相容時，解析出更寬的 Schema 併產生 Schema 變更事件。 static：僅在作業啟動時進行一次Schema解析，後續根據初始Schema解析資料，不會產生Schema變更事件。 說明 Schema解析詳情可見Kafka SQL連接器。 僅VVR 8.0.11及以上版本支援該配置項。
  scan.max.pre.fetch.records	Schema初始解析時，對每個分區最多嘗試消費解析的訊息數量	否	Int	50	在作業實際讀取並處理資料前，對每個分區嘗試提前消費指定數量的最新訊息，用於初始化Schema資訊。
  key.fields-prefix	自訂添加到訊息鍵（Key）解析出欄位名稱的首碼，以避免Kafka訊息鍵解析後的命名衝突問題。	否	String	無	假設該配置項設為key_，當key中包含欄位名a時，解析key後該欄位名稱為key_a。 說明 key.fields-prefix的配置值不可以是value.fields-prefix的首碼。
  value.fields-prefix	自訂添加到訊息體（Value）解析出欄位名稱的首碼，以避免Kafka訊息體解析後的命名衝突問題。	否	String	無	假設該配置項設為value_，當value中包含欄位名b時，解析value後該欄位名稱為value_b。 說明 value.fields-prefix的配置值不可以是key.fields-prefix的首碼。
  metadata.list	需要傳遞給下遊的中繼資料列	否	String	無	可用的中繼資料列包括topic、partition、offset、timestamp、timestamp-type、headers、leader-epoch、__raw_key__和__raw_value__。使用英文逗號分隔。 說明 __raw_key__和 __raw_value__中繼資料列需要VVR 11.6及更高版本可用。
  scan.value.initial-schemas.ddls	通過DDL方式指定某些表的初始Schema。	否	String	無	多個DDL使用英文;串連。例如，使用CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT);為表db1.t1和表db1.t2分別指定初始Schema。 這裡的DDL表結構應當和寫入目標表保持一致，並且滿足Flink SQL的文法規則。 說明 VVR 11.5及以上版本支援該配置。
  ingestion.ignore-errors	是否忽略資料解析過程中的報錯。	否	Boolean	false	說明 VVR 11.5及以上版本支援該配置。
  ingestion.error-tolerance.max-count	忽略資料解析過程報錯時，累計多少次報錯後作業失敗。	否	Integer	-1	僅當ingestion.ignore-errors開啟時生效，預設值-1表示解析異常不觸發作業失敗。 說明 VVR 11.5及以上版本支援該配置。
  scan.duplicate-field.strategy	在 Key 和 Value 部分解析出重複欄位名時，應該如何進行處理。	否	String	EXCEPTION	參數取值如下： EXCEPTION：在 key 和 value 存在重複欄位時，拋出異常。這是 VVR 11.6 及更早版本的預設行為。 PREFER_KEY：在欄位重複時，優先選擇 key 欄位的值。 PREFER_VALUE：在欄位重複時，優先選擇 value 欄位的值。 說明 VVR 11.7 及以上版本支援該配置。

  • 源表 Debezium JSON 格式

    參數	是否必填	資料類型	預設值	描述
    debezium-json.distributed-tables	否	Boolean	false	如果Debezium JSON內單張表資料會出現在多個分區，則需要開啟此選項。 說明 僅VVR 8.0.11及以上版本支援該配置項。 重要 修改該配置項後，需要無狀態啟動作業。
    debezium-json.schema-include	否	Boolean	false	設定Debezium Kafka Connect時，可以啟用Kafka配置value.converter.schemas.enable，以在訊息中包含schema。此選項表明Debezium JSON訊息是否包含schema。 參數取值如下： true：Debezium JSON訊息包含schema。 false：Debezium JSON訊息不包含schema。
    debezium-json.ignore-parse-errors	否	Boolean	false	參數取值如下： true：當解析異常時，跳過當前行。 false（預設值）：報出錯誤，作業啟動失敗。
    debezium-json.infer-schema.primitive-as-string	否	Boolean	false	解析表結構時，是否解析所有類型為String類型。 參數取值如下： true：解析所有基本類型為String。 false（預設值）：按照基本規則進行解析。
    debezium-json.infer-schema.string-type-inference	否	Boolean	true	是否嘗試將字串欄位推斷為 TIME、DATE 或 TIMESTAMP類型。設定為 false 時跳過該推斷，欄位保持 STRING。 說明 VVR 11.8及以上版本支援該配置。

  • 源表 Canal JSON 格式

    參數	是否必填	資料類型	預設值	描述
    canal-json.distributed-tables	否	Boolean	false	如果Canal JSON內單張表資料會出現在多個分區，則需要開啟此選項。 說明 僅VVR 8.0.11及以上版本支援該配置項。 重要 修改該配置項後，需要無狀態啟動作業。
    canal-json.database.include	否	String	無	一個可選的Regex，通過正則匹配Canal記錄中的database元欄位，僅讀取指定資料庫的changelog記錄。正則字串與Java的Pattern相容。
    canal-json.table.include	否	String	無	一個可選的Regex，通過正則匹配Canal記錄中的table元欄位，僅讀取指定表的changelog記錄。正則字串與Java的Pattern相容。
    canal-json.ignore-parse-errors	否	Boolean	false	參數取值如下： true：當解析異常時，跳過當前行。 false（預設值）：報出錯誤，作業啟動失敗。
    canal-json.infer-schema.primitive-as-string	否	Boolean	false	解析表結構時，是否所有類型解析為String類型。 參數取值如下： true：解析所有基本類型為String。 false（預設值）：按照基本規則進行解析。
    canal-json.infer-schema.strategy	否	String	AUTO	解析表結構時的解析策略。 參數取值如下： AUTO（預設值）：通過解析JSON資料自動解析。如果資料中不包含sqlType欄位，建議使用AUTO以避免解析失敗。 SQL_TYPE：通過canal json資料中的sqlType數組解析。如果資料包含sqlType欄位時，建議將canal-json.infer-schema.strategy設定為SQL_TYPE以獲得更精確的類型。 MYSQL_TYPE：通過canal json資料中的mysqlType數組解析。 當Kafka中的Canal JSON資料包含sqlType欄位且需要更精確的類型映射時，建議將canal-json.infer-schema.strategy設定為SQL_TYPE。 sqlType類型映射規則請見Kafka SQL連接器。 說明 VVR 11.1及以上版本支援該配置。 MYSQL_TYPE在VVR 11.3及以上版本支援使用。
    canal-json.mysql.treat-mysql-timestamp-as-datetime-enabled	否	Boolean	true	是否將mysql timestamp類型映射到cdc timestamp類型： true（預設）：mysql timestamp類型映射到cdc timestamp類型。 false：mysql timestamp類型映射到cdc timestamp_ltz類型。
    canal-json.mysql.treat-tinyint1-as-boolean.enabled	否	Boolean	true	使用MYSQL_TYPE解析時，是否將mysql tinyint(1)類型映射到cdc boolean類型： true（預設）：mysql tinyint(1)類型映射到cdc boolean類型。 false：mysql tinyint(1)類型映射到cdc tinyint(1)類型。 該配置僅當canal-json.infer-schema.strategy配置為MYSQL_TYPE時生效。
    canal-json.infer-schema.string-type-inference	否	Boolean	true	是否嘗試將字串欄位推斷為 TIME、DATE 或 TIMESTAMP類型。設定為 false 時跳過該推斷，欄位保持 STRING。 說明 VVR 11.8及以上版本支援該配置。

  • 源表 JSON 格式

    參數	是否必填	資料類型	預設值	描述
    json.timestamp-format.standard	否	String	SQL	指定輸入和輸出時間戳記格式。參數取值如下： SQL：解析yyyy-MM-dd HH:mm:ss.s{precision}格式的輸入時間戳記，例如2020-12-30 12:13:14.123。 ISO-8601：解析yyyy-MM-ddTHH:mm:ss.s{precision}格式的輸入時間戳記，例如2020-12-30T12:13:14.123。
    json.ignore-parse-errors	否	Boolean	false	參數取值如下： true：當解析異常時，跳過當前行。 false（預設值）：報出錯誤，作業啟動失敗。
    json.infer-schema.primitive-as-string	否	Boolean	false	解析表結構時，是否解析所有類型為String類型。 參數取值如下： true：解析所有基本類型為String。 false（預設值）：按照基本規則進行解析。
    json.infer-schema.flatten-nested-columns.enable	否	Boolean	false	解析JSON格式資料時，是否遞迴式地展開JSON中的嵌套列。參數取值如下： true：遞迴式展開。 false（預設值）：將嵌套列當作String處理。
    json.decode.parser-table-id.fields	否	String	無	解析JSON格式資料時，是否使用部分JSON欄位值產生tableId，多個欄位使用英文,串連。例如：JSON資料為{"col0":"a", "col1","b", "col2","c"}，產生結果如下： 配置 tableId col0 a col0,col1 a.b col0,col1,col2 a.b.c
    json.infer-schema.fixed-types	否	String	無	解析JSON格式資料時，指定某些欄位的具體類型，多個欄位使用英文,串連。例如：id BIGINT, name VARCHAR(10)可以指定JSON資料中的id欄位類型為BIGINT，name欄位類型為VARCHAR(10)。 說明 VVR 11.5及以上版本支援該配置。 VVR 11.5版本使用該配置時，需要額外添加配置scan.max.pre.fetch.records: 0。
    json.decode.converter-class	否	String	無	在解析JSON資料前，通過converter可以對JSON byte[]提前進行修改，填寫具體實作類別的全限定名。 說明 VVR 11.6及以上版本支援該配置。
    json.decode.empty-value-as-delete.enabled	否	Boolean	false	是否將Kafka compacted topic中的墓碑訊息（value 為空白）解析為 DELETE 事件，用於 compacted topic 鏡像、CDC 刪除訊號等以空 value 表示刪除語義的情境。 說明 VVR 11.7 及以上版本支援該配置。
    json.infer-schema.string-type-inference	否	Boolean	true	是否嘗試將字串欄位推斷為 TIME、DATE 或 TIMESTAMP類型。設定為 false 時跳過該推斷，欄位保持 STRING。 說明 VVR 11.8及以上版本支援該配置。

• 結果表

  參數	說明	是否必填	資料類型	預設值	備忘
  type	目標端類型。	是	String	無	固定值為kafka
  name	目標端名稱。	否	String	無	無
  topic	Kafka Topic名稱。	否	String	無	開啟時，所有的資料都會寫入這個Topic。 說明 如果沒有開啟，每條資料會寫入到其TableID對應字串（通過.拼接產生）的Topic，例如databaseName.tableName。
  partition.strategy	資料寫入Kafka分區的策略。	否	String	all-to-zero	取值如下： all-to-zero(預設值):將所有資料寫入 0 號分區。 hash-by-key:根據主鍵的雜湊值將資料寫到多個分區。保證同一個主鍵的資料在同一個分區並且有序。
  sink.tableId-to-topic.mapping	上遊表名到下遊 Kafka Topic名的映射關係。	否	String	無	每個映射關係由;分割，上遊表的表名和下遊 Kafka 的Topic名由:分割，表名可以使用Regex，映射到同一個topic的多張表可以使用,拼接。例如mydb.mytable1:topic1;mydb.mytable2:topic2。 說明 配置這個參數能夠在保留原始表名資訊的同時修改映射的Topic。

  • 結果表Canal JSON格式

    參數	是否必填	資料類型	預設值	描述
    canal-json.serialize.update.keep-changed-fields-only	否	Boolean	false	寫入 canal-json 格式的 UPDATE 訊息時，old 部分是否僅包含發生變化欄位的變更前的值。 說明 VVR 11.8及以上版本支援該配置。

  • 結果表Debezium JSON格式

    參數	是否必填	資料類型	預設值	描述
    debezium-json.include-schema.enabled	否	Boolean	false	Debezium JSON資料中是否包含Schema資訊。
    debezium-json.emit.full-table-id.enabled	否	Boolean	false	是否將完整的三段式Table ID寫入Debezium JSON中繼資料欄位中。 啟用此參數時的映射關係為： CDC Table ID部分 Debezium JSON鍵 Namespace db Schema schema Table table 禁用此參數時的映射關係為： CDC Table ID部分 Debezium JSON鍵 Namespace 無 Schema db Table table 說明 VVR 11.6及以上版本支援此配置。

source:
  type: kafka
  using.built-in-catalog: kafka_catalog

讀取單獨topic

讀取topic customers，寫入阿里雲資料湖的配置樣本：

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous

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

其中，Json格式預設產生的表名和topic名稱相同。

讀取多個topic

讀取符合Regex的多個topic，寫入 StarRocks SQL連接器 的配置樣本：

source:
  type: kafka
  topic-pattern: user_event_.*
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous

sink:
  type: starrocks
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}
 
  # （可選）資料量不大的作業建議調低 flush 間隔，避免資料長時間不落盤（預設 300000，即 5 分鐘）
  sink.buffer-flush.interval-ms: 5000
  # （可選）上遊為 utf8mb4 字元集時建議設為 4，避免文本截斷（預設 3）
  unicode-char.max-bytes: 4
  # （可選）自動建表的分桶數；StarRocks 2.5.7 以下必須顯式配置，更高版本可自動推斷
  table.create.num-buckets: 8
  # （可選）自動建表的副本數，按叢集情況配置
  table.create.properties.replication_num: 3
  # （可選）StarRocks 3.2 及以上建議開啟，加速表結構變更
  table.create.properties.fast_schema_evolution: true
  # 注意：通過 transform 變更主鍵時，必須同時設定 sink.ignore.update-before: false，
  # 否則舊主鍵對應的行會殘留在下遊

其中，Json格式預設產生的表名和topic名稱相同。

讀取key並防止欄位衝突

為了防止 key 和 value 的欄位衝突導致報錯，有以下解決方案：

1. 為欄位添加首碼，避免欄位衝突：

source:
  type: kafka
  topic: ${kafka.topic}
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  key.format: json
  value.format: json
  # key部分的欄位名添加key_首碼
  key.fields-prefix: key_
  # value部分的欄位名添加value_首碼
  value.fields-prefix: value_
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous

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

2. 配置衝突解決方案策略，詳情請見Kafka SQL連接器。如下配置會優先使用key中欄位，出現重複時忽略value中的同名欄位。

source:
  type: kafka
  topic: ${kafka.topic}
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  key.format: json
  value.format: json
  # 優先使用key的欄位，忽略同名的value中欄位
  scan.duplicate-field.strategy: PREFER_KEY
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous

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

增加中繼資料列

讀取topic customers，並在欄位增加中繼資料topic和partition的配置樣本：

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous
  metadata.list: topic,partition

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

解析報錯處理

資料解析報錯會導致作業失敗，支援配置容忍解析報錯，常配合髒資料收集使用。

完全忽略解析報錯的配置樣本：

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous
  # 開啟忽略解析報錯，預設忽略全部解析報錯
  ingestion.ignore-errors: 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
  
# 開啟髒資料收集器，列印解析失敗資料
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

累計解析失敗30次後作業報錯的配置樣本：

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous
  # 開啟忽略解析報錯
  ingestion.ignore-errors: true
  # 解析報錯發生30次後觸發作業失敗
  ingestion.error-tolerance.max-count: 30
  
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
  
# 開啟髒資料收集器，列印解析失敗資料
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

JSON資料讀取

接下來介紹 Json 格式讀取常見的一些使用方式。

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous
  # 指定欄位 db 和 table 作為 table id
  value.json.decode.parser-table-id.fields: db,tbl

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
  
# 開啟髒資料收集器，列印解析失敗資料
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # 指定欄位 db 和 tbl 作為 table id
  value.json.decode.parser-table-id.fields: db,tbl
  # 固定指定部分欄位類型
  value.json.infer-schema.fixed-types: 'db STRING, tbl STRING, id BIGINT, amount DECIMAL(18, 2)'
  # 允許未聲明欄位繼續動態推斷
  schema.inference.strategy: continuous

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
  
# 開啟髒資料收集器，列印解析失敗資料
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Canal Json資料讀取

接下來介紹 Canal Json 格式讀取常見的一些使用方式。

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: canal-json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous

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
  
# 開啟髒資料收集器，列印解析失敗資料
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: canal-json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous
  # 使用 mysql type 資訊推導 Schema，也可以配置為 SQL_TYPE 通過 sql type 推導
  value.canal-json.infer-schema.strategy: MYSQL_TYPE

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
  
# 開啟髒資料收集器，列印解析失敗資料
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Debezium Json資料讀取

讀取topic customers，寫入阿里雲資料湖的配置樣本：

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: debezium-json
  # （可選）動態識別每條資料Schema，並比對產生Schema變更
  schema.inference.strategy: continuous

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

MySQL 原始 binlog 同步 Kafka

Flink CDC資料攝入支援同步MySQL 原始binlog內容到 canal json，如下作業可以同步多張表的binlog到 topic order_dw_tables。

source:
  type: mysql
  hostname: #{hostname}
  port: 3306
  username: #{username}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  #（可選）同步增量階段新建立的表的資料
  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
  # 在 Canal JSON 中補充 mysqlType、sqlType、sql、isDdl 等資訊
  include-binlog-meta.enable: true
  
sink:
  type: kafka
  properties.bootstrap.servers: localhost:9092
  topic: order_dw_tables
  # Kafka value 使用 Canal JSON changelog 格式
  value.format: canal-json
  # 指定序列化日期類型資料時使用的格式
  value.canal-json.timestamp-format.standard: SQL
  # 資料統一寫入分區0，保證binlog順序
  partition.strategy: all-to-zero

表結構解析和變更同步策略

Kafka連接器會維護當前已知的所有表的Schema。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 使用資料中的 db、table 欄位作為 Table ID
  json.decode.parser-table-id.fields: db,table
  # 設定初始表結構
  scan.value.initial-schemas.ddls: CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT);

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 設定特定欄位始終為固定類型
  json.infer-schema.fixed-types: id BIGINT, name VARCHAR(10)
  scan.max.pre.fetch.records: 0

髒資料容忍與收集

在某些情況下，您的 Kafka 資料來源中可能包含一些格式不正確的資料（髒資料），為了避免同步作業因為這些髒資料導致頻繁失敗重啟，您可以配置忽略這部分異常資料。配置樣本如下：

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 開啟髒資料容忍功能
  ingestion.ignore-errors: true
  # 容忍 1000 條髒資料
  ingestion.error-tolerance.max-count: 1000

這裡配置了忽略 1000 條髒資料的策略，讓您的作業能夠在存在小批量髒資料時正常運行，而在髒資料條數超過這個閾值時讓作業進入失敗狀態，提醒您對資料進行校正。

如果您希望作業永遠不因為髒資料的存在導致失敗，可以採用如下的配置：

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 開啟髒資料容忍功能
  ingestion.ignore-errors: true
  # 容忍所有的髒資料
  ingestion.error-tolerance.max-count: -1

髒資料容忍策略保障了作業不會因為異常資料頻繁失敗，您可能還希望進一步的瞭解這些髒資料的資訊以調整 Kafka 資料生產者的行為，參考髒資料收集中介紹的流程，您可以在 TaskManager 日誌中查看到作業的髒資料，配置樣本如下：

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 開啟髒資料容忍功能
  ingestion.ignore-errors: true
  # 容忍所有的髒資料
  ingestion.error-tolerance.max-count: -1

pipeline:
  dirty-data.collector:
    # 將髒資料寫入 TaskManager 的記錄檔中
    type: logger

表名與Topic的映射策略

在使用kafka作為資料攝入作業的目標端時，由於寫入到Kafka訊息格式（debezium-json或者canal-json）中還包含表名資訊，後續消費Kafka訊息時往往以資料中的表名資訊作為實際表名（而非topic名稱），因此需要謹慎配置表名與Topic的映射策略。

假設在MySQL有mydb.mytable1,mydb.mytable2兩張表需要同步，可能的配置策略有以下幾種：

1. 不配置任何映射策略

在沒有任何映射策略的情況下，每張表會寫入到對應的由“庫名.表名”組成的topic中。因此mydb.mytable1的資料會寫入到名為mydb.mytable1的topic中，mydb.mytable2的資料會寫入到名為mydb.mytable2的topic中。配置樣本如下：

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: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

2. 配置route規則進行映射（不推薦）

在很多情境下，使用者不希望寫入的topic直接為“庫名.表名”的格式，希望將資料寫入到指定的topic中，因此會配置route規則進行映射。配置樣本如下：

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: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  
 route:
  - source-table: mydb.mytable1,mydb.mytable2
    sink-table: mytable1

此時所有來自mydb.mytable1,mydb.mytable2的資料都會寫入到mytable1這一個topic中。

然而，通過route規則修改寫入的topic名稱時，也會修改Kafka訊息（debezium-json或者canal-json格式）中的表名資訊，此時Kafka訊息中所有的表名都為mytable1，在其他系統消費這個topic的Kafka訊息時，可能出現不符合預期的情況。

3. 配置sink.tableId-to-topic.mapping參數進行映射（推薦）

為了在配置表名與Topic的映射規則的同時保留源表表名資訊，可以使用sink.tableId-to-topic.mapping參數完成該需求。配置樣本如下：

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: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  sink.tableId-to-topic.mapping: mydb.mytable1,mydb.mytable2:mytable

或者

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: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  sink.tableId-to-topic.mapping: mydb.mytable1:mytable;mydb.mytable2:mytable

此時所有來自mydb.mytable1,mydb.mytable2的資料都會寫入到mytable1這一個topic中，並且Kafka訊息（debezium-json或者canal-json格式）中的表名資訊仍然為mydb.mytable1或者mydb.mytable2，在其他系統消費這個topic的Kafka訊息時，能正確擷取源表表名資訊。

JSON Converter實現

JSON格式的資料有時無法保證格式完全一致，或者有處理需求無法通過Transform配置模組完成，例如合并兩個列為新的列並刪除原有的兩列，並保證能進行 Schema 變更同步。為了更靈活處理JSON資料，可以通過實現KafkaPayloadConverter介面在架構處理前，提前對JSON資料進行修改，使用時需要添加配置json.decode.converter-class填寫具體實作類別的全限定名。

實現和使用JSON Converter的流程如下：

1. 實現KafkaPayloadConverter介面並打包，在開源Demo倉庫已經提供了一些實現樣本。

2. 將打好的包添加到資料攝入作業的附加依賴。

3. 在Source模組填寫json.decode.converter-class配置，以開源專案中的ArrayElementExtractorConverter類為例：

   source:                                                                                                                                                                                                                                                                                                            
     type: kafka                                                                                                                                                                                                                                                                                          
     properties.bootstrap.servers: localhost:9092                                                                                                                                                                                                                                                                     
     topic: my_cdc_topic                                                                                                                                                                                                                                                                                              
     properties.group.id: flink-cdc-group                                                                                                                                                                                                                                                                             
     scan.startup.mode: earliest-offset  
     value.format: json                                                                                                                                                                                                                                                                              
     # 對 Kafka 訊息 value 的 JSON 資料應用自訂轉換器
     value.json.decode.converter-class: org.apache.flink.cdc.connectors.kafka.ArrayElementExtractorConverter

4. 部署作業並運行。

key format和value format

在VVR 11.5及以前的版本中，key format 和 value format的配置無法進行區分，當使用相同的format時，format配置會同時應用在key format和value format。

VVR 11.6及以後的版本對這部分進行了最佳化，以json format 為例，format 的配置項的路由規則如下：

• format首碼（如json.infer-schema.primitive-as-string）： 為了保持版本相容，format首碼的配置會預設同時應用到 key format和 value format。

• key首碼加format首碼（如key.json.infer-schema.primitive-as-string）： key首碼加format首碼的配置會只應用在key format，優先順序比format首碼的配置更高。

• value首碼加format首碼（如value.json.infer-schema.primitive-as-string）： value首碼加format首碼的配置會只應用在value format，優先順序比format首碼的配置更高。

如下的Kafka source配置，為key format和value format配置了不同的JSON Converter。

source:                                                                                                                                                                                                                                                                                                            
  type: kafka                                                                                                                                                                                                                                                                                          
  properties.bootstrap.servers: localhost:9092                                                                                                                                                                                                                                                                     
  topic: my_cdc_topic                                                                                                                                                                                                                                                                                              
  properties.group.id: flink-cdc-group                                                                                                                                                                                                                                                                             
  scan.startup.mode: earliest-offset 
  key.format: json  
  value.format: json                                                                                                                                                                                                                                                                            
  # 對 Kafka 訊息 key 的 JSON 資料應用自訂轉換器
  key.json.decode.converter-class: com.example.KeyExampleConverter                                                                                                                                                                                                                                                                            
  # 對 Kafka 訊息 value 的 JSON 資料應用自訂轉換器
  value.json.decode.converter-class: com.example.ValueExampleConverter