Realtime Compute for Apache Flink：訊息佇列Kafka

背景資訊

前提條件

注意事項

目前不推薦使用事務寫入，這是 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	否	無	說明 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'
);

Datastream API

• 構建Kafka Source

  Kafka Source提供了構建類來建立Kafka Source的執行個體。我們將通過以下範例程式碼介紹如何構建Kafka Source來消費input-topic最早位點的資料，消費組名稱為my-group，並將Kafka訊息體還原序列化為字串。

  Java

  KafkaSource<String> source = KafkaSource.<String>builder()
      .setBootstrapServers(brokers)
      .setTopics("input-topic")
      .setGroupId("my-group")
      .setStartingOffsets(OffsetsInitializer.earliest())
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build();
  
  env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

  在構建KafkaSource時，必須指定以下參數。

  參數	說明
  BootstrapServers	Kafka Broker地址，通過setBootstrapServers(String)方法配置。
  GroupId	消費者組ID，通過setGroupId(String)方法配置。
  Topics或Partition	訂閱的Topic或Partition名稱。Kafka Source提供了以下三種Topic或Partition的訂閱者式： Topic列表，訂閱Topic列表中所有Partition。 KafkaSource.builder().setTopics("topic-a","topic-b") Regex匹配，訂閱與Regex所匹配的Topic下的所有Partition。 KafkaSource.builder().setTopicPattern("topic.*") Partition列表，訂閱指定的Partition。 final HashSet<TopicPartition> partitionSet = new HashSet<>(Arrays.asList( new TopicPartition("topic-a", 0), // Partition 0 of topic "topic-a" new TopicPartition("topic-b", 5))); // Partition 5 of topic "topic-b" KafkaSource.builder().setPartitions(partitionSet)
  Deserializer	解析Kafka訊息的還原序列化器。 還原序列化器通過setDeserializer(KafkaRecordDeserializationSchema)來指定，其中KafkaRecordDeserializationSchema定義了如何解析Kafka的ConsumerRecord。如果只解析Kafka訊息中的訊息體（Value）的資料，可以通過以下任何一種方式實現： 使用Flink提供的KafkaSource構建類中的setValueOnlyDeserializer(DeserializationSchema)方法，其中DeserializationSchema定義了如何解析Kafka訊息體中的位元據。 使用Kafka提供的解析器，包括多種實作類別。例如，可以使用StringDeserializer來將Kafka訊息體解析成字串。 import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); 說明 如果需要完整地解析ConsumerRecord，則需要自行實現KafkaRecordDeserializationSchema介面。

  XML

  Maven中央庫中已經放置了Kafka DataStream連接器。

  <dependency>
      <groupId>com.alibaba.ververica</groupId>
      <artifactId>ververica-connector-kafka</artifactId>
      <version>${vvr-version}</version>
  </dependency>

  在使用Kafka DataStream連接器時，需要瞭解以下Kafka屬性：

  • 起始消費位點

    Kafka Source能夠通過位點初始化器（OffsetsInitializer）來指定從不同的位移量開始消費。內建的位點初始化器包括以下內容。

    位點初始化器	代碼設定
    從最早位點開始消費。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    從最末尾位點開始消費。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    從時間戳記大於等於指定時間的資料開始消費，單位為毫秒。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    從消費組提交的位點開始消費，如果提交位點不存在，使用最早位點。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    從消費組提交的位點開始消費，不指錨點重設策略。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    說明

    • 如果以上內建的初始化器不能滿足需求，也可以自己實現自訂的位點初始化器。

    • 如果未指錨點初始化器，則預設使用OffsetsInitializer.earliest()，即最早位點。

  • 流模式和批模式

    Kafka Source支援流式和批式兩種運行模式。預設情況下，Kafka Source設定為以流模式運行，因此作業永遠不會停止，直到Flink作業失敗或被取消。如果要配置Kafka Source在批模式下運行，可以使用setBounded(OffsetsInitializer)指定停止位移量，當所有分區都達到其停止位移量時，Kafka Source會退出運行。

    說明

    通常，流模式下Kafka Source沒有停止位移量。為了方便對代碼進行調試，流模式下可以使用 setUnbounded(OffsetsInitializer) 指定停止位移量。請留意指定流模式和批模式停止位移量的方法名（setUnbounded 和 setBounded）是不同的。

  • 動態分區檢查

    為了在不重啟Flink作業的情況下，處理Topic擴容或建立Topic等情境，可以在提供的Topic或Partition訂閱模式下，啟用動態分區檢查功能。DataStream連接器中，該功能預設關閉，需要手動顯式配置開啟：

    KafkaSource.builder()
        .setProperty("partition.discovery.interval.ms", "10000") // 每10秒檢查一次新分區。

    重要

    動態分區檢查功能依賴於Kafka叢集的元資訊更新機制。如果Kafka叢集未及時更新分區資訊，可能導致新分區無法被發現。請確保Kafka叢集的partition.discovery.interval.ms配置與實際情況匹配。

  • 事件時間和浮水印

    Kafka Source預設使用Kafka訊息中的時間戳記作為事件時間。可以自訂浮水印策略（Watermark Strategy）以從訊息中提取事件時間，並向下遊發送浮水印。

    env.fromSource(kafkaSource, new CustomWatermarkStrategy(), "Kafka Source With Custom Watermark Strategy")

    如果需要瞭解自訂浮水印策略（Watermark Strategy），請參見Generating Watermarks。

    說明

    若並行Source的部分任務長期處於空閑狀態（如某個Kafka分區長時間無資料輸入，或Source並發數超過Kafka分區數量），可能導致Watermark產生機制失效。此時系統將無法正常觸發視窗計算，造成資料處理流程停滯。

    為解決此類問題，可通過以下方式調整：

    • 配置Watermark逾時機制：啟用table.exec.source.idle-timeout參數，強制系統在指定逾時時間後產生Watermark，確保視窗計算周期的推進。

    • 資料來源最佳化：建議保持Kafka分區數與Source並發度的合理比例（推薦分區數≥Source並行度）。

  • 消費位點提交

    Kafka Source在Checkpoint完成時，提交當前的消費位點，以保證Flink的Checkpoint狀態和Kafka Broker上的提交位點一致。如果未開啟Checkpoint，Kafka Source依賴於Kafka Consumer內部的位點定時自動認可邏輯，自動認可功能由enable.auto.commit和 auto.commit.interval.ms兩個Kafka Consumer配置項進行配置。

    說明

    Kafka Source不依賴於Broker上提交的位點來恢複失敗的作業。提交位點只是為了上報Kafka Consumer和消費組的消費進度，以在Broker端進行監控。

  • 其他屬性

    除了上述屬性之外，還可以使用setProperties(Properties) 和setProperty(String, String) 為Kafka Source和Kafka Consumer設定任意屬性。KafkaSource通常有以下配置項。

    配置項	說明
    client.id.prefix	指定用於Kafka Consumer的用戶端ID首碼。
    partition.discovery.interval.ms	定義Kafka Source檢查新分區的時間間隔。 說明 partition.discovery.interval.ms會在批模式下被覆蓋為-1。
    register.consumer.metrics	指定是否在Flink中註冊Kafka Consumer的指標。
    其他Kafka Consumer配置	Kafka Consumer的配置詳情，請參見Apache Kafka。 重要 Kafka Connector會強制覆蓋部分手動設定的參數項，覆蓋詳情如下： key.deserializer始終被覆蓋為ByteArrayDeserializer。 value.deserializer始終被覆蓋為ByteArrayDeserializer。 auto.offset.reset.strategy被覆蓋為OffsetsInitializer#getAutoOffsetResetStrategy()。

    以下樣本展示如何配置Kafka Consumer，以使用PLAIN作為SASL機制並提供JAAS配置。

    KafkaSource.builder()
        .setProperty("sasl.mechanism", "PLAIN")
        .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";")

  • 監控

    Kafka Source在Flink中註冊指標，用於監控和診斷。

    • 指標範圍

      Kafka source reader的所有指標都註冊在KafkaSourceReader指標組下，KafkaSourceReader是operator指標組的子組。與特定主題分區相關的指標註冊在KafkaSourceReader.topic.<topic_name>.partition.<partition_id>指標組中。

      例如Topic "my-topic"、分區1的當前消費位點（currentOffset）註冊在<some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset。成功提交位點的次數（commitsSucceeded）註冊在<some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded。

    • 指標列表

      指標名稱	描述	範圍
      currentOffset	當前消費位點	TopicPartition
      committedOffset	當前提交位點	TopicPartition
      commitsSucceeded	成功提交的次數	KafkaSourceReader
      commitsFailed	失敗的提交次數	KafkaSourceReader

    • Kafka Consumer指標

      Kafka Consumer的指標註冊在指標組KafkaSourceReader.KafkaConsumer。例如Kafka Consumer指標records-consumed-total註冊在<some_parent_groups>.operator.KafkaSourceReader.KafkaConsumer.records-consumed-total。

      可以使用配置項register.consumer.metrics配置是否註冊Kafka消費者的指標。預設此選項設定為true。對於Kafka Consumer的指標，可參見Apache Kafka。

• 構建Kafka Sink

  Flink Kafka Sink可以實現將流資料寫入一個或多個Kafka Topic。

  DataStream<String> stream = ...
  
  
  Properties properties = new Properties();
  properties.setProperty("bootstrap.servers", );
  KafkaSink<String> kafkaSink =
                  KafkaSink.<String>builder()
                          .setKafkaProducerConfig(kafkaProperties) // // producer config
                          .setRecordSerializer(
                                  KafkaRecordSerializationSchema.builder()
                                          .setTopic("my-topic") // target topic
                                          .setKafkaValueSerializer(StringSerializer.class) // serialization schema
                                          .build())
                          .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // fault-tolerance
                          .build();
  
  stream.sinkTo(kafkaSink);

  需要配置以下參數。

  參數	說明
  Topic	資料寫入的預設Topic名稱。
  資料序列化	構建時需要提供KafkaRecordSerializationSchema來將輸入資料轉換為 Kafka的ProducerRecord。Flink提供了schema構建器，以提供一些通用的組件，例如訊息鍵（key）/訊息體（value）序列化、topic選擇、訊息分區，同樣也可以通過實現對應的介面來進行更豐富的控制。ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp)方法會在每條資料流入的時候調用，來產生ProducerRecord寫入Kafka。 使用者可以對每條資料如何寫入Kafka進行細粒度地控制。通過ProducerRecord可以進行以下操作： 設定寫入的Topic名稱。 定義訊息鍵（Key）。 指定資料寫入的Partition。
  Kafka用戶端屬性	bootstrap.servers必填，以逗號分隔的Kafka Broker列表。
  容錯語義	啟用Flink的Checkpoint後，Flink Kafka Sink可以保證精確一次的語義。除了啟用Flink的Checkpoint外，還可以通過DeliveryGuarantee參數來指定不同的容錯語義，DeliveryGuarantee參數詳情如下： DeliveryGuarantee.NONE：（預設設定）Flink不做任何保證，資料可能會丟失或重複。 DeliveryGuarantee.AT_LEAST_ONCE：保證不會丟失任何資料，但可能會重複。 DeliveryGuarantee.EXACTLY_ONCE：使用Kafka事務提供精確一次的語義保證。 說明 使用EXACTLY_ONCE語義時，需要注意的事項請參見EXACTLY_ONCE語義注意事項。

髒資料容忍與收集

在某些情況下，您的 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.tableId-to-topic.mapping: mydb.mytable1,mydb.mytable2:mytable

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

或者

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.tableId-to-topic.mapping: mydb.mytable1:mytable;mydb.mytable2:mytable

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

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

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叢集？

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

• 空間管理與操作