AI 助理
备案控制台
文档
输入文档关键字查找
首页 实时计算 Flink版 开发参考 连接器 Upsert Kafka

Upsert Kafka

更新时间：
产品详情
相关技术圈
我的收藏

本文为您介绍如何使用Upsert Kafka连接器。

背景信息

Upsert Kafka连接器支持以upsert方式从Kafka topic中读取数据并将数据写入Kafka topic。

  • 作为源表，此连接器可以将Kafka中存储的数据转换为changelog流，其中每条数据记录代表一个更新或删除事件。更准确地说，数据记录中的value被解释为同一key的最后一个value的UPDATE，如果有这个key，如果不存在相应的key，则该更新被视为INSERT。用表来类比，changelog流中的数据记录被解释为UPSERT，也称为INSERT或UPDATE，因为任何具有相同key的现有行都被覆盖。另外，value为空的消息将会被视作为DELETE消息。

  • 作为结果表或数据摄入目标端，此连接器可以消费上游计算逻辑产生的changelog流。它会将INSERT或UPDATE_AFTER数据作为正常的Kafka消息写入，并将DELETE数据以value为空的Kafka消息写入，表示对应key的消息被删除。Flink将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新或删除消息将落在同一分区中。

类别

详情

支持类型

源表和结果表，数据摄入目标端

运行模式

流模式

数据格式

avro、avro-confluent、csv、json和raw

特有监控指标

  • 源表

    • numRecordsIn

    • numRecordsInPerSecond

    • numBytesIn

    • numBytesInPerScond

    • currentEmitEventTimeLag

    • currentFetchEventTimeLag

    • sourceIdleTime

    • pendingRecords

  • 结果表

    • numRecordsOut

    • numRecordsOutPerSecond

    • numBytesOut

    • numBytesOutPerSecond

    • currentSendTime

API种类

SQL，数据摄入YAML作业

是否支持更新或删除结果表数据

是

前提条件

使用限制

  • 仅Flink计算引擎VVR 2.0.0及以上版本支持消息队列Kafka连接器。

  • 仅支持读取和写入Apache Kafka 0.10及以上版本的数据。

  • 仅支持Apache Kafka 2.8版本的客户端配置项，详情请参见Apache Kafka消费者和生产者配置项文档。

  • Upsert Kafka结果表在使用精确一次语义时，写入的Kafka集群必须开启事务功能，且仅支持Apache Kafka 0.11及以上版本的集群。

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}

配置项

参数

说明

数据类型

是否必填

默认值

备注

type

目标端类型。

STRING

是

无

固定值为upsert-kafka。

name

目标端名称。

STRING

否

无

无。

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。

sink.delivery-guarantee

写入时的语义模式。

STRING

否

at-least-once

取值如下：

  • none：不保证任何语义，数据可能会丢失或重复。

  • at-least-once（默认值）：保证数据不丢失，但可能会重复。

  • exactly-once：使用Kafka事务保证数据不会丢失和重复。

sink.add-tableId-to-header-enabled

是否将table信息写入header。

BOOLEAN

否

false

开启时，namespace、schemaName和tableName会分别写入header。

aliyun.kafka.accessKeyId

阿里云账号AccessKey ID。

STRING

否

无

详情请参见创建AccessKey。

说明

同步数据到阿里云消息队列Kafka版时需要配置。

aliyun.kafka.accessKeySecret

阿里云账号AccessKey Secret。

STRING

否

无

详情请参见创建AccessKey。

说明

同步数据到阿里云消息队列Kafka版时需要配置。

aliyun.kafka.instanceId

阿里云Kafka消息队列实例ID。

STRING

否

无

请在阿里云Kafka实例详情界面查看。

说明

同步数据到阿里云消息队列Kafka版时需要配置。

aliyun.kafka.endpoint

阿里云Kafka API服务接入地址。

STRING

否

无

详情请参见服务接入点。

说明

同步数据到阿里云消息队列Kafka版时需要配置。

aliyun.kafka.regionId

Topic所在实例的地域ID。

STRING

否

无

详情请参见服务接入点。

说明

同步数据到阿里云消息队列Kafka版时需要配置。

使用示例

  • 源表

    创建Kafka数据源表，源表中包含网站用户的浏览数据。

    CREATE TABLE pageviews(
user_id BIGINT,
page_id BIGINT,
viewtime TIMESTAMP,
user_region STRING,
WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
)WITH(
'connector'='kafka',
'topic'='<yourTopicName>',
'properties.bootstrap.servers'='...',
'format'='json'
);

  • 结果表

    • 创建Upsert Kafka结果表。

      CREATE TABLE pageviews_per_region(
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'
);

    • 将统计网站用户的浏览数据写入结果表中。

      INSERT INTO pageviews_per_region
SELECT
user_region,
COUNT(*),
COUNT(DISTINCTuser_id)
FROM pageviews
GROUP BY user_region;

  • 数据摄入目标端

    source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: upsert-kafka
  name: Upsert Kafka Sink
  properties.bootstrap.servers: ${upsert.kafka.bootstraps.server}
  aliyun.kafka.accessKeyId: ${upsert.kafka.aliyun.ak}
  aliyun.kafka.accessKeySecret: ${upsert.kafka.aliyun.sk}
  aliyun.kafka.instanceId: ${upsert.kafka.aliyun.instanceid}
  aliyun.kafka.endpoint: ${upsert.kafka.aliyun.endpoint}
  aliyun.kafka.regionId: ${upsert.kafka.aliyun.regionid}

route:
  - source-table: ${mysql.source.table}
    sink-table: ${upsert.kafka.topic}

最佳实践

上一篇：云数据库Redis下一篇：Elasticsearch
文档推荐