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}

配置項