Realtime Compute for Apache Flink：Kafka SQL連接器

背景資訊

前提條件

注意事項

目前不推薦使用事務寫入，這是 Flink 社區和 Kafka 社區的設計缺陷所致。當設定sink.delivery-guarantee = exactly-once，Kafka Connector 會啟用事務寫入，存在三個已知問題：

• 每個 Checkpoint 會產生一個 Transaction ID。如果 Checkpoint 間隔太短，Transaction ID會過多。Kafka 叢集的 Coordinator 可能因此記憶體不足，從而破壞 Kafka 叢集的穩定性。

• 每個事務會建立一個 Producer 執行個體。如果同時提交的事務太多，TaskManager 的記憶體可能耗盡，從而破壞 Flink 作業的穩定性。

• 多個 Flink 作業若使用相同的sink.transactional-id-prefix，它們產生的事務 ID 可能衝突。一個作業寫入失敗時，會阻塞 Kafka 分區的 LSO（Log Start Offset）前進，這會影響所有消費者讀取該分區的資料。

如果你需要 Exactly-Once 語義，改用 Upsert Kafka 寫入主鍵表，並用主鍵保證等冪性。如果需要使用事務寫入，請參見EXACTLY_ONCE語義注意事項。

網路連接排查

Flink作業啟動時若報錯Timed out waiting for a node assignment，通常是因為 Flink 與 Kafka 之間網路不通。

SQL

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

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

元資訊列

CREATE TABLE kafka_source (
  --讀取訊息所屬的topic作為`record_topic`欄位
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  --讀取ConsumerRecord中的時間戳記作為`ts`欄位
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  --讀取訊息的offset作為`record_offset`欄位
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  --將`ts`欄位中的時間戳記作為ProducerRecord的時間戳記寫入Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

WITH參數

• 通用

  參數	說明	資料類型	是否必填	預設值	備忘
  connector	表類型。	String	是	無	固定值為kafka。
  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。 不能通過該方式修改以下配置，因為它們會被Kafka連接器覆蓋： key.deserializer value.deserializer
  format	讀取或寫入Kafka訊息value部分時使用的格式。	String	否	無	支援的格式 csv json avro debezium-json canal-json maxwell-json avro-confluent raw 說明 更多format參數設定請參見Format參數。
  key.format	讀取或寫入Kafka訊息key部分時使用的格式。	String	否	無	支援的格式 csv json avro debezium-json canal-json maxwell-json avro-confluent raw 說明 使用該配置時，key.options配置是必填的。
  key.fields	Kafka訊息key部分對應的源表或結果表欄位。	String	否	無	多個欄位名以分號（;）分隔。例如field1;field2
  key.fields-prefix	為所有Kafka訊息key部分指定自訂首碼，以避免與訊息value部分格式欄位重名。	String	否	無	該配置項僅用於源表和結果表的列名區分，解析和產生Kafka訊息key部分時，該首碼會被移除。 說明 使用該配置時，value.fields-include必須配置為EXCEPT_KEY。
  value.format	讀取或寫入Kafka訊息value部分時使用的格式。	String	否	無	該配置等同於format，只能設定 format 或 value.format 中的一個。如果同時配置，value.format會覆蓋format。
  value.fields-include	在解析或產生Kafka訊息value部分時，是否要包含訊息key部分對應的欄位。	String	否	ALL	參數取值如下： ALL（預設值）：所有列都會作為Kafka訊息value部分處理 EXCEPT_KEY：除去key.fields定義的欄位，剩餘欄位作為Kafka訊息value部分處理

• 源表

  參數	說明	資料類型	是否必填	預設值	備忘
  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	否	KafkaSource-{源表表名}	如果指定的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的所有分區資料。 說明 在Realtime Compute引擎VVR 6.0.x版本中，動態分區檢測預設為關閉。自8.0版本起該功能預設開啟，活動訊號間隔時間預設設定為5分鐘。
  scan.header-filter	根據Kafka資料是否包含指定的訊息頭（Header）對資料進行條件過濾。	String	否	無	Header key和value使用冒號（:）分隔，多個header條件之間使用邏輯運算子（&、|）串連，支援取反邏輯運算子（!）。例如depart:toy|depart:book&!env:test表示保留header中包含depart=toy或depart=book，且不包含env=test的Kafka資料。 說明 僅Realtime Compute引擎VVR 8.0.6及以上版本支援配置該參數。 暫不支援括弧運算。 邏輯運算順序為從左至右。 Header value會以UTF-8格式轉換為字串，與參數指定的header value進行比較。
  scan.check.duplicated.group.id	是否檢查通過properties.group.id指定的消費者組有重複。	Boolean	否	false	參數取值如下： true：在啟動作業前，系統會檢查消費者組是否存在重複。若發現重複，作業將報錯並停止運行，從而避免與現有消費者組產生衝突。 false：直接啟動作業，不檢查消費者組衝突。 說明 僅VVR 6.0.4及以上版本支援該參數。

• 結果表

  參數	說明	資料類型	是否必填	預設值	備忘
  topic	寫入的topic名稱。	String	是	無	無
  sink.partitioner	從Flink並發到Kafka分區的映射模式。	String	否	default	取值如下： default（預設值）：使用Kafka預設的分區模式 fixed：每個Flink並發對應一個固定的Kafka分區。 round-robin：Flink並發中的資料將被輪流分配至Kafka的各個分區。 自訂分區映射模式：如果fixed和round-robin不滿足需求，可以建立一個FlinkKafkaPartitioner的子類來自訂分區映射模式。例如org.mycompany.MyPartitioner
  sink.delivery-guarantee	Kafka結果表的語義模式。	String	否	at-least-once	取值如下： none：不保證任何語義，資料可能會丟失或重複。 at-least-once（預設值）：保證資料不丟失，但可能會重複。 exactly-once：使用Kafka事務保證資料不會丟失和重複。 說明 在使用exactly-once語義時，sink.transactional-id-prefix是必填的。
  sink.transactional-id-prefix	在exactly-once語義下使用的Kafka事務ID首碼。	String	否	無	只有sink.delivery-guarantee配置為exactly-once時該配置才會生效。
  sink.parallelism	Kafka結果表運算元的並發數。	Integer	否	無	上遊運算元的並發，由架構決定。

安全與認證

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /*SSL配置*/
  /*佈建服務端提供的truststore (CA 憑證) 的路徑*/
  /*檔案管理上傳的檔案會存放在/flink/usrlib/路徑下*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*如果要求用戶端認證，則需要配置keystore (私密金鑰) 的路徑*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*用戶端驗證伺服器位址的演算法，空值表示禁用伺服器位址驗證*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*SASL配置*/
  /*將SASL機制配置為as SCRAM-SHA-256*/
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /*配置JAAS*/
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

樣本中提到的CA認證和私密金鑰可使用Realtime Compute控制台的檔案管理功能上傳至平台。上傳後檔案存放在/flink/usrlib目錄下，需要使用的CA認證檔案名稱為my-truststore.jks，則上傳後在WITH參數中有兩種方式可以設定'properties.ssl.truststore.location'來使用該認證：

• 配置'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'，採用這種方式Flink運行期間不需要動態下載OSS檔案，但是不支援偵錯模式。

• Realtime Compute引擎版本為 VVR 11.5 及以上，可以配置properties.ssl.truststore.location和properties.ssl.keystore.location為OSS絕對路徑地址，檔案路徑格式為oss://flink-fullymanaged-<工作空間ID>/artifacts/namespaces/<專案空間名稱>/<檔案名稱>。採用這種方式會在Flink運行期間動態下載OSS檔案，支援偵錯模式。

源表起始位點

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  --從最早位點開始消費
  'scan.startup.mode' = 'earliest-offset',
  --從最末尾位點開始消費
  'scan.startup.mode' = 'latest-offset',
  --從消費者組"my-group"的已提交位點開始消費
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- 如果 "my-group" 為首次使用，則從最早位點開始消費
  'properties.auto.offset.reset' = 'latest', -- 如果 "my-group" 為首次使用，則從最末尾位點開始消費
  --從指定的毫秒時間戳記1655395200000開始消費
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  --從指錨點開始消費
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

源表位點提交

Kafka源表只在checkpoint成功後將當前消費位點提交至Kafka叢集。如果checkpoint間隔設定較長，在Kafka叢集側觀察到的消費位點會有延遲。在進行checkpoint時，Kafka源表會將當前讀取進度儲存在狀態中，並不依賴於提交到叢集上的位點進行故障恢複，提交位點僅僅是為了在Kafka側能夠監控到讀取進度，位點提交失敗不會對資料正確性產生任何影響。

結果表自訂分區器

如果內建的Kafka Producer分區模式無法滿足需求，可以實現自訂分區模式將資料寫入對應的分區。自訂分區器需要繼承FlinkKafkaPartitioner，開發完成後編譯JAR包，使用檔案管理功能上傳至Realtime Compute控制台。上傳並引用完成後，請在WITH參數中設定sink.partitioner參數，參數值為分區器完整的類路徑，如org.mycompany.MyPartitioner。

Kafka、Upsert Kafka或Kafka JSON catalog的選擇

Kafka是一種只能添加資料的訊息佇列系統，無法進行資料的更新和刪除操作，因此在流式SQL計算中無法處理上遊的CDC變更資料和彙總、聯合等運算元的回撤邏輯。如果需要將含有變更或回撤類型的資料寫入Kafka，請使用對變更資料進行特殊處理的Upsert Kafka結果表。

為了方便將上遊資料庫中一個或多個資料表中的變更資料批量同步到Kafka中，可以使用Kafka JSON catalog。如果Kafka中儲存的資料格式為JSON，使用Kafka JSON catalog可以省去定義schema和WITH參數的步驟。詳情可參見管理Kafka JSON Catalog。

使用樣本

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    --可選，將嵌套列全部展開。
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
--使用COALESCE處理空值情況。

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "阿里雲",
    "engine": "Flink"
  }
}

CREATE TEMPORARY TABLE kafka_source (
  id          VARCHAR,
  `name`      VARCHAR,
  properties  ROW<`owner` STRING, engine STRING>
) WITH (
  'connector' = 'kafka',
  'topic' = 'xxx',
  'properties.bootstrap.servers' = 'xxx',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

EXACTLY_ONCE語義注意事項

• 配置消費者隔離等級

  所有消費 Kafka 資料的應用必須設定 isolation.level：

  • read_committed：唯讀取已提交的資料。

  • read_uncommitted（預設）：可讀取未提交的資料。

  EXACTLY_ONCE 依賴read_committed。否則消費者可能看到未提交資料，破壞一致性。

• 事務逾時與資料丟失

  Flink 從 Checkpoint 恢複時，僅依賴該 Checkpoint 開始前已提交的事務。如果作業崩潰到重啟的時間超過 Kafka 事務逾時，Kafka 會自動中止事務，導致資料丟失。

  • Kafka Broker 預設transaction.max.timeout.ms = 15 分鐘。

  • Flink Kafka Sink 預設設定 transaction.timeout.ms = 1 小時。

  • 你必須在 Broker 端提高transaction.max.timeout.ms，使其不小於 Flink 的設定。

• Producer 池與 Checkpoint 並發

  EXACTLY_ONCE 模式使用固定大小的 Kafka Producer 池。每個 Checkpoint 佔用池中的一個 Producer。如果並發 Checkpoint 數超過池大小，作業會失敗。

  請根據最大並發 Checkpoint 數調整 Producer 池大小。

• 並行度縮容限制

  如果作業在第一個 Checkpoint 前失敗，重啟後不會保留原有 Producer 池資訊。因此，在第一個 Checkpoint 完成前，不要縮減作業並行度。如必須縮容，並行度不得低於FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR。

• 事務阻塞讀取

  在read_committed模式下，任何未結束（未提交也未中止）的事務會阻塞整個 Topic 的讀取。

  例如：

  • 事務 1 寫入資料。

  • 事務 2 寫入並提交資料。

  • 只要事務 1 未結束，事務 2 的資料對消費者不可見。

  因此：

  • 正常運行時，資料可見延遲約等於 Checkpoint 間隔。

  • 作業失敗時，正在寫入的 Topic 會阻塞消費者，直到作業重啟或事務逾時，極端情況下，事務逾時甚至會影響讀取。

常見問題

• Flink如何擷取Kafka中的JSON資料？

• 為什麼Kafka源表資料基於Event Time的視窗後，不能輸出資料？

• 如何串連配置了安全資訊的Kafka叢集？

• 如何解決欄位命名衝突的問題？

• 空間管理與操作