Realtime Compute for Apache Flink：Upsert Kafka

SQL

Upsert Kafka連接器支援以upsert方式從Kafka topic中讀取資料並將資料寫入Kafka topic。

文法結構

CREATE TABLE upsert_kafka_sink(
user_region STRING,
pv BIGINT,
uv BIGINT,
PRIMARY KEY(user_region) NOT ENFORCED
)WITH(
'connector'='upsert-kafka',
'topic'='<yourTopicName>',
'properties.bootstrap.servers'='...',
'key.format'='avro',
'value.format'='avro'
);

WITH參數

• 通用

  參數	說明	資料類型	是否必填	預設值	備忘
  connector	表類型。	String	是	無	固定值為upsert-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
  key.format	讀取或寫入Kafka訊息key部分時使用的格式。	String	是	無	當使用該配置時，key.fields或key.fields-prefix配置是必填的。 參數取值如下： csv json avro debezium-json canal-json maxwell-json avro-confluent raw
  key.fields-prefix	為所有Kafka訊息key部分指定自訂首碼，以避免與訊息value部分格式欄位重名。	String	否	無	該配置項僅用於源表和結果表的列名區分，解析和產生Kafka訊息key部分時，該首碼會被移除。 說明 使用該配置時，value.fields-include必須配置為EXCEPT_KEY。
  value.format	讀取或寫入Kafka訊息value部分時使用的格式。	String	是	無	該配置等同於format，因此format和value.format 只能配置其中一個，如果同時配置兩個會產生衝突。
  value.fields-include	在解析或產生Kafka訊息value部分時，是否要包含訊息key部分對應的欄位。	String	是	ALL	參數取值如下： ALL（預設值）：所有列都會作為Kafka訊息value部分處理。 EXCEPT_KEY：除去key.fields定義的欄位，剩餘欄位作為Kafka訊息value部分處理。
  topic	讀取或寫入topic名稱。	String	是	無	無。

• 結果表

  參數	說明	資料類型	是否必填	預設值	備忘
  sink.parallelism	Kafka結果表運算元的並發數。	Integer	否	上遊運算元的並發，由架構決定	無。
  sink.buffer-flush.max-rows	緩衝重新整理前，最多能緩衝多少條記錄。	Integer	否	0（未開啟）	當結果表收到很多同key上的更新時，緩衝將保留同key的最後一條記錄，因此結果表緩衝能協助減少發往Kafka topic的資料量，以及避免發送潛在的tombstone訊息。 說明 如果要開啟結果表緩衝，需要同時設定sink.buffer-flush.max-rows和sink.buffer-flush.interval兩個選項為大於零的值。
  sink.buffer-flush.interval	緩衝重新整理的間隔時間。	Duration	否	0（未開啟）	單位可以為毫秒（ms）、秒（s）、分鐘（min）或小時（h）。例如'sink.buffer-flush.interval'='1 s'。 當結果表收到很多同key上的更新時，緩衝將保留同key的最後一條記錄，因此結果表緩衝能協助減少發往Kafka topic的資料量，以及避免發送潛在的tombstone訊息。 說明 如果要開啟結果表緩衝，需要同時設定sink.buffer-flush.max-rows和sink.buffer-flush.interval兩個選項為大於零的值。

資料攝入

Upsert Kafka連接器可以用於資料攝入YAML作業開發，作為目標端寫入。寫入時使用JSON格式，主鍵欄位也會放入訊息體中。

文法結構

sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: localhost:9092
  # 阿里雲訊息佇列 Kafka 版
  aliyun.kafka.accessKeyId: ${secret_values.kafka-ak}
  aliyun.kafka.accessKeySecret: ${secret_values.kafka-sk}
  aliyun.kafka.instanceId: ${instancd-id}
  aliyun.kafka.endpoint: ${endpoint}
  aliyun.kafka.regionId: ${region-id}

配置項

支援的類型變更

資料攝入Upsert Kafka連接器支援全部類型的變更操作，但是在讀取時需要使用Schema固定的Flink Upsert Kafka SQL連接器進行讀取。