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连接器进行读取。