背景信息

前提條件

使用限制

CREATE TABLE AS（CTAS）的使用限制

• 僅Flink計算引擎vvr-4.0.12-flink-1.13及以上版本支持Kafka作為CREATE TABLE AS（CTAS）的同步數(shù)據(jù)源。

• 僅支持JSON格式的類型推導(dǎo)和schema變更，其它數(shù)據(jù)格式暫不支持。

• 僅支持Kafka中value部分的類型推導(dǎo)和表結(jié)構(gòu)變更。如果您需要同步Kafka key部分的列，則需要您手動在DDL中進(jìn)行指定。詳情請參見示例三。

數(shù)據(jù)攝入YAML的使用限制

• 僅支持Debezium JSON和Canal JSON格式的類型推導(dǎo)和表結(jié)構(gòu)變更，其他數(shù)據(jù)格式暫不支持。

• 對于數(shù)據(jù)源，僅支持同一張表的數(shù)據(jù)在同一分區(qū)，不支持單表數(shù)據(jù)分布在多個分區(qū)。

網(wǎng)絡(luò)連接排查

如果您的Flink作業(yè)在啟動時出現(xiàn)Timed out waiting for a node assignment錯誤，一般是Flink和Kafka之間的網(wǎng)絡(luò)連通問題導(dǎo)致的。

如果Kafka服務(wù)端沒有正確配置，客戶端在第一步收到的連接地址有誤，即使properties.bootstrap.servers配置的地址可以連接上，也無法正常讀取或?qū)懭霐?shù)據(jù)。該問題經(jīng)常在Flink與Kafka之間存在代理、端口轉(zhuǎn)發(fā)、專線等網(wǎng)絡(luò)轉(zhuǎn)發(fā)機(jī)制時發(fā)生。

SQL

Kafka連接器可以在SQL作業(yè)中使用，作為源表或者結(jié)果表。

語法結(jié)構(gòu)

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參數(shù)

• 通用

  參數(shù)	說明	數(shù)據(jù)類型	是否必填	默認(rèn)值	備注
  connector	表類型。	String	是	無	固定值為kafka。
  properties.bootstrap.servers	Kafka broker地址。	String	是	無	格式為host:port,host:port,host:port，以英文逗號（,）分割。
  properties.*	對Kafka客戶端的直接配置。	String	否	無	后綴名必須是Kafka官方文檔中定義的生產(chǎn)者和消費(fèi)者配置。 Flink會將properties.前綴移除，并將剩余的配置傳遞給Kafka客戶端。例如可以通過'properties.allow.auto.create.topics'='false'來禁用自動創(chuàng)建topic。 不能通過該方式修改以下配置，因?yàn)樗鼈儠籏afka連接器覆蓋： key.deserializer value.deserializer
  format	讀取或?qū)懭隟afka消息value部分時使用的格式。	String	否	無	支持的格式 csv json avro debezium-json canal-json maxwell-json avro-confluent raw 說明 更多format參數(shù)設(shè)置請參見Format參數(shù)。
  key.format	讀取或?qū)懭隟afka消息key部分時使用的格式。	String	否	無	支持的格式 csv json avro debezium-json canal-json maxwell-json avro-confluent raw 說明 使用該配置時，key.options配置是必填的。
  key.fields	Kafka消息key部分對應(yīng)的源表或結(jié)果表字段。	String	否	無	多個字段名以分號（;）分隔。例如field1;field2
  key.fields-prefix	為所有Kafka消息key部分指定自定義前綴，以避免與消息value部分格式字段重名。	String	否	無	該配置項(xiàng)僅用于源表和結(jié)果表的列名區(qū)分，解析和生成Kafka消息key部分時，該前綴會被移除。 說明 使用該配置時，value.fields-include必須配置為EXCEPT_KEY。
  value.format	讀取或?qū)懭隟afka消息value部分時使用的格式。	String	否	無	該配置等同于format，因此format和value.format 只能配置其中一個，如果同時配置兩個會產(chǎn)生沖突
  value.fields-include	在解析或生成Kafka消息value部分時，是否要包含消息key部分對應(yīng)的字段。	String	否	ALL	參數(shù)取值如下： ALL（默認(rèn)值）：所有列都會作為Kafka消息value部分處理 EXCEPT_KEY：除去key.fields定義的字段，剩余字段作為Kafka消息value部分處理

• 源表

  參數(shù)	說明	數(shù)據(jù)類型	是否必填	默認(rèn)值	備注
  topic	讀取的topic名稱。	String	否	無	以英文分號 (;) 分隔多個topic名稱，例如topic-1和topic-2 說明 topic和topic-pattern兩個選項(xiàng)只能指定其中一個。
  topic-pattern	匹配讀取topic名稱的正則表達(dá)式。所有匹配該正則表達(dá)式的topic在作業(yè)運(yùn)行時均會被讀取。	String	否	無	說明 僅VVR 3.0.0及以上版本支持該參數(shù)。 topic和topic-pattern兩個選項(xiàng)只能指定其中一個。
  properties.group.id	消費(fèi)組ID。	String	否	KafkaSource-{源表表名}	如果指定的group id為首次使用，則必須將properties.auto.offset.reset設(shè)置為earliest或latest以指定首次啟動位點(diǎn)。
  scan.startup.mode	Kafka讀取數(shù)據(jù)的啟動位點(diǎn)。	String	否	group-offsets	取值如下： earliest-offset：從Kafka最早分區(qū)開始讀取。 latest-offset：從Kafka最新位點(diǎn)開始讀取。 group-offsets（默認(rèn)值）：從指定的properties.group.id已提交的位點(diǎn)開始讀取。 timestamp：從scan.startup.timestamp-millis指定的時間戳開始讀取。 specific-offsets：從scan.startup.specific-offsets指定的偏移量開始讀取。 說明 該參數(shù)在作業(yè)無狀態(tài)啟動時生效。作業(yè)在從checkpoint重啟或狀態(tài)恢復(fù)時，會優(yōu)先使用狀態(tài)中保存的進(jìn)度恢復(fù)讀取。
  scan.startup.specific-offsets	specific-offsets啟動模式下，指定每個分區(qū)的啟動偏移量。	String	否	無	例如partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	timestamp啟動模式下，指定啟動位點(diǎn)時間戳。	Long	否	無	單位為毫秒
  scan.topic-partition-discovery.interval	動態(tài)檢測Kafka topic和partition的時間間隔。	Duration	否	5分鐘	分區(qū)檢查間隔默認(rèn)為5分鐘。需要顯式地設(shè)置分區(qū)檢查間隔為非正數(shù)才能關(guān)閉此功能。開啟動態(tài)分區(qū)發(fā)現(xiàn)后，Kafka Source 可以自動地發(fā)現(xiàn)新增的分區(qū)并自動讀取對應(yīng)分區(qū)上的數(shù)據(jù)。在topic-pattern模式下，不僅讀取已有topic的新增分區(qū)數(shù)據(jù)，也會讀取符合正則匹配的新增topic的所有分區(qū)數(shù)據(jù)。 說明 在實(shí)時計算引擎VVR 6.0.x版本中，動態(tài)分區(qū)檢測默認(rèn)為關(guān)閉。自8.0版本起該功能默認(rèn)打開，檢測間隔默認(rèn)設(shè)置為5分鐘。
  scan.header-filter	根據(jù)Kafka數(shù)據(jù)是否包含指定的消息頭（Header）對數(shù)據(jù)進(jìn)行條件過濾。	String	否	無	Header key和value使用冒號（:）分隔，多個header條件之間使用邏輯運(yùn)算符（&、|）連接，支持取反邏輯運(yùn)算符（!）。例如depart:toy|depart:book&!env:test表示保留header中包含depart=toy或depart=book，且不包含env=test的Kafka數(shù)據(jù)。 說明 僅實(shí)時計算引擎VVR 8.0.6及以上版本支持配置該參數(shù)。 暫不支持括號運(yùn)算。 邏輯運(yùn)算順序?yàn)閺淖笾劣摇?/p> Header value會以UTF-8格式轉(zhuǎn)換為字符串，與參數(shù)指定的header value進(jìn)行比較。
  scan.check.duplicated.group.id	是否檢查通過properties.group.id指定的消費(fèi)者組有重復(fù)。	Boolean	否	false	參數(shù)取值如下： true：在啟動作業(yè)前檢查消費(fèi)者組是否有重復(fù)，如有重復(fù)作業(yè)將會報錯，避免與現(xiàn)有的消費(fèi)者組產(chǎn)生沖突。 false：直接啟動作業(yè)，不檢查消費(fèi)者組沖突。 說明 僅VVR 6.0.4及以上版本支持該參數(shù)。

• 結(jié)果表

  參數(shù)	說明	數(shù)據(jù)類型	是否必填	默認(rèn)值	備注
  topic	寫入的topic名稱。	String	是	無	無
  sink.partitioner	從Flink并發(fā)到Kafka分區(qū)的映射模式。	String	否	default	取值如下： default（默認(rèn)值）：使用Kafka默認(rèn)的分區(qū)模式 fixed：每個Flink并發(fā)對應(yīng)一個固定的Kafka分區(qū)。 round-robin：Flink并發(fā)中的數(shù)據(jù)將被輪流分配至Kafka的各個分區(qū)。 自定義分區(qū)映射模式：如果fixed和round-robin不滿足您的需求，您可以創(chuàng)建一個FlinkKafkaPartitioner的子類來自定義分區(qū)映射模式。例如org.mycompany.MyPartitioner
  sink.delivery-guarantee	Kafka結(jié)果表的語義模式。	String	否	at-least-once	取值如下： none：不保證任何語義，數(shù)據(jù)可能會丟失或重復(fù)。 at-least-once（默認(rèn)值）：保證數(shù)據(jù)不丟失，但可能會重復(fù)。 exactly-once：使用Kafka事務(wù)保證數(shù)據(jù)不會丟失和重復(fù)。 說明 在使用exactly-once語義時，sink.transactional-id-prefix是必填的。
  sink.transactional-id-prefix	在exactly-once語義下使用的Kafka事務(wù)ID前綴。	String	否	無	只有sink.delivery-guarantee配置為exactly-once時該配置才會生效。
  sink.parallelism	Kafka結(jié)果表算子的并發(fā)數(shù)。	Integer	否	無	上游算子的并發(fā)，由框架決定。

• CTAS同步數(shù)據(jù)源

  參數(shù)	說明	數(shù)據(jù)類型	是否必填	默認(rèn)值	備注
  json.infer-schema.flatten-nested-columns.enable	是否遞歸式地展開JSON中的嵌套列。	Boolean	否	false	參數(shù)取值如下： true：遞歸式展開。對于被展開的列，F(xiàn)link使用索引該值的路徑作為名字。例如，對于JSON {"nested": {"col": true}} 中的列col，它展開后的名字為nested.col。 false（默認(rèn)值）：將嵌套類型當(dāng)作String處理。
  json.infer-schema.primitive-as-string	是否推導(dǎo)所有基本類型為String類型。	Boolean	否	false	參數(shù)取值如下： true：推導(dǎo)所有基本類型為String。 false：按照基本規(guī)則進(jìn)行推導(dǎo)。

  所有Kafka consumer和producer支持的配置項(xiàng)均可在配置前添加properties.前綴后在WITH參數(shù)中使用。例如需要配置Kafka consumer或producer的超時時間request.timeout.ms為60000毫秒，則可以在WITH參數(shù)中配置'properties.request.timeout.ms'='60000'。Kafka consumer和Kafka producer的配置項(xiàng)詳情請參見Apache Kafka官方文檔。

安全與認(rèn)證

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配置*/
  /*配置服務(wù)端提供的truststore (CA 證書) 的路徑*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*如果要求客戶端認(rèn)證，則需要配置keystore (私鑰) 的路徑*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*客戶端驗(yàn)證服務(wù)器地址的算法，空值表示禁用服務(wù)器地址驗(yàn)證*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*SASL配置*/
  /*將SASL機(jī)制配置為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證書和私鑰可使用實(shí)時計算控制臺的資源上傳功能上傳至平臺，上傳后文件存放在/flink/usrlib目錄下。例如，需要使用的CA證書文件名為my-truststore.jks，則上傳后您可以在WITH參數(shù)中指定'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'來使用該證書。

源表起始位點(diǎn)

啟動模式

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

起始位點(diǎn)優(yōu)先級

在以上任何一個步驟中，如果位點(diǎn)過期或Kafka集群發(fā)生問題等原因?qū)е挛稽c(diǎn)無效，則會使用properties.auto.offset.reset指定的策略進(jìn)行位點(diǎn)重置，如果您未設(shè)置該配置項(xiàng)，則會產(chǎn)生異常要求您介入。

一種常見情況是使用全新的group id開始消費(fèi)。首先源表會向Kafka集群查詢該group的已提交位點(diǎn)，由于該group id是第一次使用，不會查詢到有效位點(diǎn)，所以會通過properties.auto.offset.reset參數(shù)配置的策略進(jìn)行重置。因此在使用全新group id進(jìn)行消費(fèi)時，必須配置properties.auto.offset.reset來指定位點(diǎn)重置策略。

源表位點(diǎn)提交

Kafka源表只在checkpoint成功后將當(dāng)前消費(fèi)位點(diǎn)提交至Kafka集群。如果您的checkpoint間隔設(shè)置較長，您在Kafka集群側(cè)觀察到的消費(fèi)位點(diǎn)會有延遲。在進(jìn)行checkpoint時，Kafka源表會將當(dāng)前讀取進(jìn)度存儲在狀態(tài)中，并不依賴于提交到集群上的位點(diǎn)進(jìn)行故障恢復(fù)，提交位點(diǎn)僅僅是為了在Kafka側(cè)能夠監(jiān)控到讀取進(jìn)度，位點(diǎn)提交失敗不會對數(shù)據(jù)正確性產(chǎn)生任何影響。

結(jié)果表自定義分區(qū)器

如果內(nèi)置的Kafka Producer分區(qū)模式無法滿足您的需求，您可以實(shí)現(xiàn)自定義分區(qū)模式將數(shù)據(jù)寫入對應(yīng)的分區(qū)。自定義分區(qū)器需要繼承FlinkKafkaPartitioner，開發(fā)完成后編譯JAR包，使用資源上傳功能上傳至實(shí)時計算控制臺。上傳并引用完成后，請在WITH參數(shù)中設(shè)置sink.partitioner參數(shù)，參數(shù)值為分區(qū)器完整的類路徑，如org.mycompany.MyPartitioner。

Kafka、Upsert Kafka或Kafka JSON catalog的選擇

Kafka是一種只能添加數(shù)據(jù)的消息隊(duì)列系統(tǒng)，無法進(jìn)行數(shù)據(jù)的更新和刪除操作，因此在流式SQL計算中無法處理上游的CDC變更數(shù)據(jù)和聚合、聯(lián)合等算子的回撤邏輯。如果您需要將含有變更或回撤類型的數(shù)據(jù)寫入Kafka，請使用對變更數(shù)據(jù)進(jìn)行特殊處理的Upsert Kafka結(jié)果表。

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

作為CTAS數(shù)據(jù)源

• 類型推導(dǎo)

  在類型推導(dǎo)過程中，F(xiàn)link默認(rèn)只展開JSON文本中的第一層數(shù)據(jù)，根據(jù)其類型和數(shù)值，按照基本規(guī)則進(jìn)行類型推導(dǎo)。類型映射基本規(guī)則如下表所示。

  JSON類型	Flink SQL類型
  BOOLEAN	BOOLEAN
  STRING	DATE、TIMESTAMP、TIMESTAMP_LTZ、TIME或 STRING
  INT或LONG	BIGINT
  BIGINT	DECIMAL或STRING 說明 Flink中DECIMAL的類型存在精度限制。因此，如果整數(shù)的實(shí)際取值超過了DECIMAL類型最大精度，F(xiàn)link會自動推導(dǎo)其類型為STRING，避免精度的損失。
  FLOAT、DOUBLE或BIG DECIMAL	DOUBLE
  ARRAY	STRING
  OBJECT	STRING

  示例

  • JSON文本

    {
      "id": 101,
      "name": "VVP",
      "properties": {
        "owner": "阿里云",
        "engine": "Flink"
      }
        "type": ["大數(shù)據(jù)"]
    }

  • Flink寫入到下游存儲的表信息為

    id	name	properties	type
    101	VVP	{ "owner": "阿里云", "engine": "Flink" }	["大數(shù)據(jù)"]

• 輔助推導(dǎo)

  當(dāng)基本規(guī)則不符合您的實(shí)際需要時，您可以在源表的DDL中聲明特定列的解析類型。通過該方式，F(xiàn)link會優(yōu)先使用您聲明的列類型去解析目標(biāo)字段。針對以下示例，F(xiàn)link會使用DECIMAL的方式去解析price字段，而不是使用默認(rèn)的基本規(guī)則將其轉(zhuǎn)換為DOUBLE類型。

  CREATE TABLE evolvingKafkaSource (
    price DECIMAL(18, 2)
  ) WITH (
    'connector' = 'kafka',
    'properties.bootstrap.servers' = 'localhost:9092',
    'topic' = 'evolving_kafka_demo',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
  );

  但是，當(dāng)您在DDL中指定的類型和實(shí)際數(shù)據(jù)中的類型不一致時，可以按照以下方式進(jìn)行處理：

  • 在聲明的類型比實(shí)際類型更寬泛時，以聲明的類型自動去解析。例如，聲明為DOUBLE，遇到的數(shù)據(jù)類型為BIGINT，則會以DOUBLE類型去解析。

  • 在實(shí)際的類型比聲明的類型更為寬泛或者兩種類型不兼容時，由于當(dāng)前CTAS不支持類型變更，因此會報錯提示您相關(guān)信息，您需要重新啟動作業(yè)并聲明準(zhǔn)確的類型去解析數(shù)據(jù)。

    類型的寬泛的程度以及兼容性如下圖所示。

    說明

    • 上圖表示越靠近根節(jié)點(diǎn)，其類型越寬泛。如果兩個類型在不同的分支上，則表示這兩個類型不兼容。

    • 不支持輔助推導(dǎo)復(fù)雜類型，包括ROW、ARRAY、MAP和MULTISET。

    • 對于復(fù)雜類型，F(xiàn)link在默認(rèn)情況下會處理為STRING。

使用示例

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處理空值情況。

Datastream API

• 構(gòu)建Kafka Source

  Kafka Source提供了構(gòu)建類來創(chuàng)建Kafka Source的實(shí)例。我們將通過以下示例代碼為您介紹如何構(gòu)建Kafka Source來消費(fèi)input-topic最早位點(diǎn)的數(shù)據(jù)，消費(fèi)組名稱為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");

  XML

  Maven中央庫中已經(jīng)放置了Kafka DataStream連接器。

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

  在構(gòu)建KafkaSource時，必須指定以下參數(shù)。

  參數(shù)	說明
  BootstrapServers	Kafka Broker地址，通過setBootstrapServers(String)方法配置。
  GroupId	消費(fèi)者組ID，通過setGroupId(String)方法配置。
  Topics或Partition	訂閱的Topic或Partition名稱。Kafka Source提供了以下三種Topic或Partition的訂閱方式： Topic列表，訂閱Topic列表中所有Partition。 KafkaSource.builder().setTopics("topic-a","topic-b") 正則表達(dá)式匹配，訂閱與正則表達(dá)式所匹配的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）的數(shù)據(jù)，則您可以通過以下任何一種方式實(shí)現(xiàn)： 使用Flink提供的KafkaSource構(gòu)建類中的setValueOnlyDeserializer(DeserializationSchema)方法，其中DeserializationSchema定義了如何解析Kafka消息體中的二進(jìn)制數(shù)據(jù)。 使用Kafka提供的解析器，包括多種實(shí)現(xiàn)類。例如，您可以使用StringDeserializer來將Kafka消息體解析成字符串。 import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); 說明 如果需要完整地解析ConsumerRecord，則需要您自己實(shí)現(xiàn)KafkaRecordDeserializationSchema接口。

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

  • 起始消費(fèi)位點(diǎn)

    Kafka Source能夠通過位點(diǎn)初始化器（OffsetsInitializer）來指定從不同的偏移量開始消費(fèi)。內(nèi)置的位點(diǎn)初始化器包括以下內(nèi)容。

    位點(diǎn)初始化器	代碼設(shè)置
    從最早位點(diǎn)開始消費(fèi)。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    從最末尾位點(diǎn)開始消費(fèi)。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    從時間戳大于等于指定時間的數(shù)據(jù)開始消費(fèi)，單位為毫秒。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    從消費(fèi)組提交的位點(diǎn)開始消費(fèi)，如果提交位點(diǎn)不存在，使用最早位點(diǎn)。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    從消費(fèi)組提交的位點(diǎn)開始消費(fèi)，不指定位點(diǎn)重置策略。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    說明

    • 如果以上內(nèi)置的初始化器不能滿足需求，您也可以自己實(shí)現(xiàn)自定義的位點(diǎn)初始化器。

    • 如果您未指定位點(diǎn)初始化器，則默認(rèn)使用OffsetsInitializer.earliest()，即最早位點(diǎn)。

  • 流模式和批模式

    Kafka Source支持流式和批式兩種運(yùn)行模式。默認(rèn)情況下，Kafka Source設(shè)置為以流模式運(yùn)行，因此作業(yè)永遠(yuǎn)不會停止，直到Flink作業(yè)失敗或被取消。如果要配置Kafka Source在批模式下運(yùn)行，您可以使用setBounded(OffsetsInitializer)指定停止偏移量，當(dāng)所有分區(qū)都達(dá)到其停止偏移量時，Kafka Source會退出運(yùn)行。

    說明

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

  • 動態(tài)分區(qū)檢查

    為了在不重啟Flink作業(yè)的情況下，處理Topic擴(kuò)容或新建Topic等場景，您可以在提供的Topic或Partition訂閱模式下，啟用動態(tài)分區(qū)檢查功能。

    說明

    默認(rèn)開啟動態(tài)分區(qū)檢查功能，分區(qū)檢查間隔默認(rèn)為5分鐘。需要顯式地設(shè)置分區(qū)檢查間隔為非正數(shù)才能關(guān)閉此功能。代碼示例如下。

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

  • 事件時間和水印

    Kafka Source默認(rèn)使用Kafka消息中的時間戳作為事件時間。您可以自定義水印策略（Watermark Strategy）以從消息中提取事件時間，并向下游發(fā)送水印。

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

    如果您需要了解自定義水印策略（Watermark Strategy），請參見Generating Watermarks。

  • 消費(fèi)位點(diǎn)提交

    Kafka Source在Checkpoint完成時，提交當(dāng)前的消費(fèi)位點(diǎn)，以保證Flink的Checkpoint狀態(tài)和Kafka Broker上的提交位點(diǎn)一致。如果未開啟Checkpoint，Kafka Source依賴于Kafka Consumer內(nèi)部的位點(diǎn)定時自動提交邏輯，自動提交功能由enable.auto.commit和 auto.commit.interval.ms兩個Kafka Consumer配置項(xiàng)進(jìn)行配置。

    說明

    Kafka Source不依賴于Broker上提交的位點(diǎn)來恢復(fù)失敗的作業(yè)。提交位點(diǎn)只是為了上報Kafka Consumer和消費(fèi)組的消費(fèi)進(jìn)度，以在Broker端進(jìn)行監(jiān)控。

  • 其他屬性

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

    配置項(xiàng)	說明
    client.id.prefix	指定用于Kafka Consumer的客戶端ID前綴。
    partition.discovery.interval.ms	定義Kafka Source檢查新分區(qū)的時間間隔。 說明 partition.discovery.interval.ms會在批模式下被覆蓋為-1。
    register.consumer.metrics	指定是否在Flink中注冊Kafka Consumer的指標(biāo)。
    其他Kafka Consumer配置	Kafka Consumer的配置詳情，請參見Apache Kafka。 重要 Kafka Connector會強(qiáng)制覆蓋部分您手動配置的參數(shù)項(xiàng)，覆蓋詳情如下： key.deserializer始終被覆蓋為ByteArrayDeserializer。 value.deserializer始終被覆蓋為ByteArrayDeserializer。 auto.offset.reset.strategy被覆蓋為OffsetsInitializer#getAutoOffsetResetStrategy()。

    以下示例為您展示如何配置Kafka Consumer，以使用PLAIN作為SASL機(jī)制并提供JAAS配置。

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

  • 監(jiān)控

    Kafka Source在Flink中注冊指標(biāo)，用于監(jiān)控和診斷。

    • 指標(biāo)范圍

      Kafka source reader的所有指標(biāo)都注冊在KafkaSourceReader指標(biāo)組下，KafkaSourceReader是operator指標(biāo)組的子組。與特定主題分區(qū)相關(guān)的指標(biāo)注冊在KafkaSourceReader.topic.<topic_name>.partition.<partition_id>指標(biāo)組中。

      例如Topic "my-topic"、分區(qū)1的當(dāng)前消費(fèi)位點(diǎn)（currentOffset）注冊在<some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset。成功提交位點(diǎn)的次數(shù)（commitsSucceeded）注冊在<some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded。

    • 指標(biāo)列表

      指標(biāo)名稱	描述	范圍
      currentOffset	當(dāng)前消費(fèi)位點(diǎn)	TopicPartition
      committedOffset	當(dāng)前提交位點(diǎn)	TopicPartition
      commitsSucceeded	成功提交的次數(shù)	KafkaSourceReader
      commitsFailed	失敗的提交次數(shù)	KafkaSourceReader

    • Kafka Consumer指標(biāo)

      Kafka Consumer的指標(biāo)注冊在指標(biāo)組KafkaSourceReader.KafkaConsumer。例如Kafka Consumer指標(biāo)records-consumed-total注冊在<some_parent_groups>.operator.KafkaSourceReader.KafkaConsumer.records-consumed-total。

      您可以使用配置項(xiàng)register.consumer.metrics配置是否注冊Kafka消費(fèi)者的指標(biāo)。默認(rèn)此選項(xiàng)設(shè)置為true。對于Kafka Consumer的指標(biāo)，您可以參見Apache Kafka。

• 構(gòu)建Kafka Sink

  Flink Kafka Sink可以實(shí)現(xiàn)將流數(shù)據(jù)寫入一個或多個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);

  您需要配置以下參數(shù)。

  參數(shù)	說明
  Topic	數(shù)據(jù)寫入的默認(rèn)Topic名稱。
  數(shù)據(jù)序列化	構(gòu)建時需要提供KafkaRecordSerializationSchema來將輸入數(shù)據(jù)轉(zhuǎn)換為 Kafka 的ProducerRecord。Flink 提供了 schema 構(gòu)建器 以提供一些通用的組件，例如消息鍵（key）/消息體（value）序列化、topic 選擇、消息分區(qū)，同樣也可以通過實(shí)現(xiàn)對應(yīng)的接口來進(jìn)行更豐富的控制。ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp)方法會在每條數(shù)據(jù)流入的時候調(diào)用，來生成ProducerRecord寫入Kafka。 用戶可以對每條數(shù)據(jù)如何寫入Kafka進(jìn)行細(xì)粒度地控制。通過ProducerRecord可以進(jìn)行以下操作： 設(shè)置寫入的Topic名稱。 定義消息鍵（Key）。 指定數(shù)據(jù)寫入的Partition。
  Kafka客戶端屬性	bootstrap.servers必填，以逗號分隔的Kafka Broker列表。
  容錯語義	啟用Flink的Checkpoint后，F(xiàn)link Kafka Sink可以保證精確一次的語義。除了啟用Flink的Checkpoint外，您還可以通過DeliveryGuarantee參數(shù)來指定不同的容錯語義，DeliveryGuarantee參數(shù)詳情如下： DeliveryGuarantee.NONE：（默認(rèn)設(shè)置）Flink不做任何保證，數(shù)據(jù)可能會丟失或重復(fù)。 DeliveryGuarantee.AT_LEAST_ONCE：保證不會丟失任何數(shù)據(jù)，但可能會重復(fù)。 DeliveryGuarantee.EXACTLY_ONCE：使用Kafka事務(wù)提供精確一次的語義保證。 說明 使用EXACTLY_ONCE語義時，需要注意的事項(xiàng)請參見EXACTLY_ONCE語義注意事項(xiàng)。

EXACTLY_ONCE語義注意事項(xiàng)

• 當(dāng)使用事務(wù)寫入Kafka時，請為所有消費(fèi)Kafka數(shù)據(jù)的應(yīng)用配置isolation.level參數(shù)。該參數(shù)取值如下：

  • read_committed：只讀取已提交的數(shù)據(jù)。

  • read_uncommitted（默認(rèn)值）：可以讀取未提交的數(shù)據(jù)。

• DeliveryGuarantee.EXACTLY_ONCE模式依賴于在從某個Checkpoint恢復(fù)后，且在該Checkpoint開始之前所提交的事務(wù)。如果Flink作業(yè)崩潰與完成重啟之間的時間大于Kafka的事務(wù)超時時間，則會有數(shù)據(jù)丟失，因?yàn)镵afka會自動中止超過超時時間的事務(wù)。因此，請根據(jù)您的預(yù)期停機(jī)時間適當(dāng)?shù)嘏渲媚氖聞?wù)超時。

  Kafka Broker默認(rèn)的transaction.max.timeout.ms設(shè)置為15分鐘，Producer設(shè)置的事務(wù)超時不能超過Broker指定的時間。Flink Kafka Sink默認(rèn)會將Kafka Producer配置中的transaction.timeout.ms屬性設(shè)置為1小時，因此在使用DeliveryGuarantee.EXACTLY_ONCE模式前，需要增加Broker端的transaction.max.timeout.ms值。

• DeliveryGuarantee.EXACTLY_ONCE模式為在Flink Kafka Producer實(shí)例中使用一個固定大小的Kafka Producer池。每個Checkpoint會使用池中的一個Kafka Producer。如果并發(fā)的Checkpoint數(shù)量超過Producer池的大小，F(xiàn)link Kafka Producer會拋出異常并使整個作業(yè)失敗。請相應(yīng)地配置Producer池大小和最大并發(fā)的Checkpoint數(shù)量。

• DeliveryGuarantee.EXACTLY_ONCE會盡可能清除阻止Consumer從Kafka topic中讀取數(shù)據(jù)的殘留事務(wù)。但如果Flink作業(yè)在第一個 Checkpoint之前就出現(xiàn)故障，則在重啟該作業(yè)后并不會保留重啟前Producer池的信息。因此，在第一個Checkpoint完成之前縮減Flink作業(yè)的并行度是不安全的，即使要縮減并行度，也不能小于FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR。

• 對于在read_committed模式下運(yùn)行的Kafka Consumer，任何未結(jié)束（既未中止也未完成）的事務(wù)都將阻塞對該Kafka Topic的所有讀取。如果您按照以下步驟進(jìn)行了操作：

  1. 用戶開啟事務(wù)1并通過該事務(wù)寫入了一些數(shù)據(jù)。

  2. 用戶開啟事務(wù)2并通過該事務(wù)寫入了更多的數(shù)據(jù)。

  3. 用戶提交事務(wù)2。

  即使來自事務(wù)2的數(shù)據(jù)已經(jīng)提交，在事務(wù)1提交或中止之前，事務(wù)2的數(shù)據(jù)對消費(fèi)者是不可見的。因此：

  • 在Flink作業(yè)正常工作期間，您可以預(yù)期寫入Kafka topic的數(shù)據(jù)會有延遲，約為Checkpoint的平均間隔。

  • 在Flink作業(yè)失敗的情況下，該作業(yè)正在寫入的Topic將會阻塞Consumer的讀取，直到作業(yè)重新啟動或事務(wù)超時。

常見問題

• Flink如何獲取JSON數(shù)據(jù)？

• Flink和Kafka網(wǎng)絡(luò)連通，但Flink無法消費(fèi)或者寫入數(shù)據(jù)？

• Flink中的Commit Offset有什么作用？

• 為什么Kafka源表數(shù)據(jù)基于Event Time的窗口后，不能輸出數(shù)據(jù)？

• 如何鏈接配置了安全信息的Kafka集群？

• 如何解決字段命名沖突的問題？

• 如何解決Flink依賴沖突問題？