搜索

Upsert Kafka Connector - 让实时统计更简单

程序源代码

共 11890字，需浏览 24分钟

 · 2021-03-18

点击上方蓝色字体，选择“设为星标”
回复”资源“获取更多资源


在某些场景中，例如读取 compacted topic 或者输出（更新）聚合结果的时候，需要将 Kafka 消息记录的 key 当成主键处理，用来确定一条数据是应该作为插入、删除还是更新记录来处理。为了实现该功能，社区为 Kafka 专门新增了一个 upsert connector（upsert-kafka），该 connector 扩展自现有的 Kafka connector，工作在 upsert 模式（FLIP-149）下。新的 upsert-kafka connector 既可以作为 source 使用，也可以作为 sink 使用，并且提供了与现有的 kafka connector 相同的基本功能和持久性保证，因为两者之间复用了大部分代码。

要使用 upsert-kafka connector，必须在创建表时定义主键，并为键（key.format）和值（value.format）指定序列化反序列化格式。

一、Upsert Kafka Connector是什么？

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

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

作为 sink，upsert-kafka 连接器可以消费 changelog 流。它会将 INSERT/UPDATE_AFTER 数据作为正常的 Kafka 消息写入，并将 DELETE 数据以 value 为空的 Kafka 消息写入（表示对应 key 的消息被删除）。Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。

如果是更新，则同一个key会存储多条数据，但在读取该表数据时，只保留最后一次更新的值)，并将 DELETE 数据以 value 为空的 Kafka 消息写入（key被打上墓碑标记，表示对应 key 的消息被删除）。

Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。

upsert-kafka connector相关参数

connector

必选。指定要使用的连接器，Upsert Kafka 连接器使用：'upsert-kafka'。

topic 必选。用于读取和写入的 Kafka topic 名称。

properties.bootstrap.servers 必选。以逗号分隔的 Kafka brokers 列表。

key.format 必选。用于对 Kafka 消息中 key 部分序列化和反序列化的格式。key 字段由 PRIMARY KEY 语法指定。支持的格式包括 'csv'、'json'、'avro'。

value.format 必选。用于对 Kafka 消息中 value 部分序列化和反序列化的格式。支持的格式包括 'csv'、'json'、'avro'。

properties 可选。该选项可以传递任意的 Kafka 参数。选项的后缀名必须匹配定义在 Kafka 参数文档中的参数名。Flink 会自动移除 选项名中的 "properties." 前缀，并将转换后的键名以及值传入 KafkaClient。例如，你可以通过 'properties.allow.auto.create.topics' = 'false' 来禁止自动创建 topic。但是，某些选项，例如'key.deserializer' 和 'value.deserializer' 是不允许通过该方式传递参数，因为 Flink 会重写这些参数的值。

value.fields-include 可选，默认为ALL。控制key字段是否出现在 value 中。当取ALL时，表示消息的 value 部分将包含 schema 中所有的字段，包括定义为主键的字段。当取EXCEPT_KEY时，表示记录的 value 部分包含 schema 的所有字段，定义为主键的字段除外。

key.fields-prefix 可选。为了避免与value字段命名冲突，为key字段添加一个自定义前缀。默认前缀为空。一旦指定了key字段的前缀，必须在DDL中指明前缀的名称，但是在构建key的序列化数据类型时，将移除该前缀。见下面的示例。在需要注意的是：使用该配置属性，value.fields-include的值必须为EXCEPT_KEY。

二、使用步骤

1.引入库

        <!-- Flink kafka connector: kafka版本大于1.0.0可以直接使用通用的连接器 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.11</artifactId>
            <version>1.12.0</version>
            <scope>provided</scope>
        </dependency>

2.SQL计算

示例：实时地统计网页PV和UV的总量

-- 创建kafka数据源表（json格式）
-- 'format.type' = 'json',                   -- required: specify the format type
-- 'format.fail-on-missing-field' = 'true',  -- optional: flag whether to fail if a field is missing or not,'false' by default
-- 'format.ignore-parse-errors' = 'true',    -- optional: skip fields and rows with parse errors instead of failing;

CREATE TABLE source_ods_fact_user_ippv (
    user_id      STRING,
    client_ip    STRING,     
    client_info  STRING,     
    pagecode     STRING,     
    access_time  TIMESTAMP,  
    dt           STRING,     
    WATERMARK FOR access_time AS access_time - INTERVAL '5' SECOND  -- 定义watermark
) WITH (
   'connector' = 'kafka',
    'topic' = 'user_ippv',
    'scan.startup.mode' = 'earliest-offset',
    'properties.group.id' = 'group1',
    'properties.bootstrap.servers' = 'xxx:9092', 
    'format' = 'json', 
    'json.fail-on-missing-field' = 'false',
    'json.ignore-parse-errors' = 'true'
);

-- 创建kafka upsert结果表且指定组合主键为：do_date,do_min
CREATE TABLE result_total_pvuv_min (
    do_date     STRING,     -- 统计日期
    do_min      STRING,      -- 统计分钟
    pv          BIGINT,     -- 点击量
    uv          BIGINT,     -- 一天内同个访客多次访问仅计算一个UV
    currenttime TIMESTAMP,  -- 当前时间
    PRIMARY KEY (do_date, do_min) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'result_total_pvuv_min',
  'properties.bootstrap.servers' = 'xxx:9092',
  'key.json.ignore-parse-errors' = 'true',
  'value.json.fail-on-missing-field' = 'false',
  'key.format' = 'json',
  'value.format' = 'json',
  'value.fields-include' = 'ALL'
);
-- 创建视图
CREATE VIEW view_total_pvuv_min AS
SELECT
     dt AS do_date,                    -- 时间分区
     count (client_ip) AS pv,          -- 客户端的IP
     count (DISTINCT client_ip) AS uv, -- 客户端去重
     max(access_time) AS access_time   -- 请求的时间
FROM
    source_ods_fact_user_ippv
GROUP BY dt;


-- 将每分钟的pv/uv统计结果写入kafka upsert表
INSERT INTO result_total_pvuv_min
SELECT
  do_date,   
  cast(DATE_FORMAT (access_time,'HH:mm') AS STRING) AS do_min,-- 分钟级别的时间
  pv,
  uv,
  CURRENT_TIMESTAMP AS currenttime
from
  view_total_pvuv_min;

该处使用示例数据和验证结果如下：

kafak 数据源：
{"user_id":"1","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1001","access_time":"2021-01-23 11:32:24","dt":"2021-01-08"}
{"user_id":"1","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1201","access_time":"2021-01-23 11:32:55","dt":"2021-01-08"}
{"user_id":"2","client_ip":"192.165.12.1","client_info":"pc","pagecode":"1031",   "access_time":"2021-01-23 11:32:59","dt":"2021-01-08"}
{"user_id":"1","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1101","access_time":"2021-01-23 11:33:24","dt":"2021-01-08"}
{"user_id":"3","client_ip":"192.168.10.3","client_info":"pc","pagecode":"1001",   "access_time":"2021-01-23 11:33:30","dt":"2021-01-08"}
{"user_id":"1","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1001","access_time":"2021-01-23 11:34:24","dt":"2021-01-08"}


实时统计的结果表(TOPIC:result_total_pvuv_min)：
{"do_date":"2021-01-08","do_min":"11:32","pv":1,"uv":1,"currenttime":"2021-01-23 08:22:06.431"}
{"do_date":"2021-01-08","do_min":"11:32","pv":2,"uv":1,"currenttime":"2021-01-23 08:22:06.526"}
{"do_date":"2021-01-08","do_min":"11:32","pv":3,"uv":2,"currenttime":"2021-01-23 08:22:06.527"}
{"do_date":"2021-01-08","do_min":"11:33","pv":4,"uv":2,"currenttime":"2021-01-23 08:22:06.527"}
{"do_date":"2021-01-08","do_min":"11:33","pv":5,"uv":3,"currenttime":"2021-01-23 08:22:06.528"}
{"do_date":"2021-01-08","do_min":"11:34","pv":6,"uv":3,"currenttime":"2021-01-23 08:22:06.529"}


----------------分割线--------------------

重测试输入如下示例数据：
{"user_id":"10","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1001","access_time":"2021-01-22 10:10:24","dt":"2021-01-22"}
{"user_id":"11","client_ip":"192.168.12.2","client_info":"phone","pagecode":"1002","access_time":"2021-01-22 11:10:24","dt":"2021-01-22"}
{"user_id":"10","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1001","access_time":"2021-01-22 10:11:24","dt":"2021-01-22"}
{"user_id":"11","client_ip":"192.168.12.3","client_info":"phone","pagecode":"1002","access_time":"2021-01-22 11:12:14","dt":"2021-01-22"}


打印待更新结果：
+----+--------------------------------+--------------------------------+----------------------+----------------------+-----------------------+
| op |                        do_date |                         do_min |                   pv |                   uv |           currenttime |
+----+--------------------------------+--------------------------------+----------------------+----------------------+-----------------------+
| +I |                     2021-01-22 |                          10:10 |                    1 |                    1 | 2021-01-23T08:33:2... |
| -U |                     2021-01-22 |                          10:10 |                    1 |                    1 | 2021-01-23T08:33:2... |
| +U |                     2021-01-22 |                          11:10 |                    2 |                    2 | 2021-01-23T08:33:2... |
| -U |                     2021-01-22 |                          11:10 |                    2 |                    2 | 2021-01-23T08:33:2... |
| +U |                     2021-01-22 |                          11:10 |                    3 |                    2 | 2021-01-23T08:33:2... |
| -U |                     2021-01-22 |                          11:10 |                    3 |                    2 | 2021-01-23T08:33:3... |
| +U |                     2021-01-22 |                          11:12 |                    4 |                    3 | 2021-01-23T08:33:3... |

3. Kafka -> FLINK -> TIDB

Flink on TIDB 在当前已经有小红书、贝壳金服等在使用，作为一个支持upsert的实时数据同步方案具备一定的可行性。

select version(); -- 5.7.25-TiDB-v4.0.8
drop table if exists result_user_behavior;
CREATE TABLE `result_user_behavior` (
  `user_id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `client_ip` varchar(30) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `client_info` varchar(30) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `page_code` varchar(30) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `access_time` TIMESTAMP COLLATE utf8mb4_general_ci DEFAULT NULL,
  `dt`varchar(30) COLLATE utf8mb4_general_ci DEFAULT NULL,
  PRIMARY KEY (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
// 支持upsert的一种可行数据同步方案
tenv.executeSql("CREATE TABLE source_kafka_user_behavior (\n" +
          "    user_id      INT,\n" +
          "    client_ip    STRING,     \n" +
          "    client_info  STRING,     \n" +
          "    page_code     STRING,     \n" +
          "    access_time  TIMESTAMP,  \n" +
          "    dt           STRING,     \n" +
          "    WATERMARK FOR access_time AS access_time - INTERVAL '5' SECOND \n" +
          ") WITH (\n" +
          "   'connector' = 'kafka',\n" +
          "    'topic' = 'user_ippv',\n" +
          "    'scan.startup.mode' = 'latest-offset',\n" +
          "    'properties.group.id' = 'test-group1',\n" +
          "    'properties.bootstrap.servers' = 'xx:9092', \n" +
          "    'format' = 'json', \n" +
          "    'json.fail-on-missing-field' = 'false',\n" +
          "    'json.ignore-parse-errors' = 'true'\n" +
          ")").print();

tenv.executeSql("CREATE TABLE sink_upsert_tidb (\n" +
      "    user_id      INT,\n" +
      "    client_ip    STRING,     \n" +
      "    client_info  STRING,     \n" +
      "    page_code     STRING,     \n" +
      "    access_time  TIMESTAMP,  \n" +
      "    dt           STRING,     \n" +
      "    PRIMARY KEY (user_id) NOT ENFORCED" +
      ") WITH (\n" +
      "  'connector' = 'jdbc',\n" +
      "  'url' = 'jdbc:mysql://xxx:4000/bi',\n" +
      "  'username' = 'bi_rw',\n" +
      "  'password' = 'xxx',\n" +
      "  'table-name' = 'result_user_behavior'\n" +
     ")");


tenv.executeSql("insert into sink_upsert_tidb" +
         " select " +
         "    user_id     ,\n" +
         "    client_ip   ,     \n" +
         "    client_info ,     \n" +
         "    page_code    ,     \n" +
         "    access_time ,  \n" +
         "    dt               \n" +
         "from source_kafka_user_behavior").print();

测试输入：

测试数据：
{"user_id":"11","client_ip":"192.168.12.3","client_info":"phone","page_code":"1002","access_time":"2021-01-25 11:12:14","dt":"2021-01-25"}
{"user_id":"11","client_ip":"192.168.12.3","client_info":"phone","page_code":"1003","access_time":"2021-01-25 11:12:14","dt":"2021-01-25"}
{"user_id":"11"} -- 值全部置空
{"user_id":"11","client_ip":"192.168.12.4","client_info":"phone","page_code":"10","access_time":"2021-01-25 11:35:14","dt":"2021-01-25"}
{"user_id":"12","client_ip":"192.168.12.5","client_info":"phone","page_code":"10","access_time":"2021-01-25 11:35:14","dt":"2021-01-25"}

Tidb查询结果示例：

总结

这里演示了使用kaka作为source和sink的使用示例，其中我们把从kafka source中消费的数据进行视图查询的时候则显示以上更新结果，每一条以统计日期和统计分钟作为联合主键的数据插入都会被解析为+I（插入）-U（标记待删除值） +U (更新新值)，这样在最新的result_total_pvuv_min 的kafka upsert 结果表中就是最新的数据。

当前kafka-upsert connector 适用于Flink-1.12的版本，作为一个数据聚合的中转对于很多业务场景有一定的普适性，比如kafka upsert结果表还可以作为维表join, 或者通过flink sink 到HDFS, iceberg table等进行离线分析。

如果想真正实时，Flink+Tidb就是一个很好的解决方案。虽然Tidb存储和计算不分离，但是能使用加机器解决的问题，性能都不是事，况且Tidb完全兼容MySQL语法，非常适合MySQL平迁，而且支持事务，和使用MySQL没有什么特别大的区别，

官方已出TiSpark查询引擎，虽还未实测性能，但想必会比MySQL 引擎查询的效率要高。我司也开始着手Tidb的使用，目前的实时的任务是基于微批的形式处理，还不能算是完全的实时，后面随着对其的了解原来越完善，完全实时化则指日可待。



FileSystem/JDBC/Kafka - Flink三大Connector实现原理及案例

企业数据治理及在美团的最佳实践


欢迎点赞+收藏+转发朋友圈素质三连

文章不错？点个【在看】吧！ 
浏览 23
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
老爸嘲讽我了，写破代码一年就挣十几万，他在工地带50个工人，一个月光人头费就3万，让我滚回去跟他干！
点击上方 "大数据肌肉猿"关注, 星标一起成长点击下方链接，进入高质量学习交流群今日更新| 1052个转型案例分享-大数据交流群来自：网络，侵删有个网友的父亲是做工程的，天天就嘲笑他，说他天天写着破代码有啥用，一年就拿个十多万的死工资，然后告诉他自己在工地里面带了50个工人，一个月能抽三万
程序源代码
0
让扩散模型听话的小秘籍？CAN：通过操控权重来控制条件生成模型，图像生成效率大升级！
↑ 点击蓝字 关注极市平台作者丨科技猛兽编辑丨极市平台极市导读 本文提出的 CAN 模型 (Condition-Aware Neural Network) 是一种对图像生成模型添加控制的方法。CAN 可以通过动态操纵神经网络的权重来控制图像生成过程。作者在 ImageNet 图像
极市平台
0
好未来测开一面，挺简单！（0428面试原题解析）
大家好，我是二哥呀。今天继续给大家分享春招面试题《好未来测开一面原题》，附详细答案，我会用通俗易懂+手绘图的方式，让天下所有的面渣都能逆袭 😁二哥的 Java 面试指南内容较长，建议正在冲刺 24 届春招和 25 届暑期实习、秋招的同学先收藏起来，面试的时候大概率会碰到，1、二哥的 Linux 速查
沉默王二
0
新规！不授予学位！博士毕业更难了？
来源：阿秒富友研究院编辑：学妹据4月22日科技日报消息，全国人大常委会法工委发言人杨合庆在近日举行的记者会上表示，即将提请十四届全国人大常委会第九次会议审议的学位法草案二次审议稿将进一步完善学位授予条件和程序，并对保障博士学位质量作出专门规定。杨合庆介绍，2023年8月，十四届全国人大常委会第五次会
机器学习初学者
0
大量 Java 开源项目停更...
点击关注公众号，Java 干货及时推送↓推荐阅读：投了 100 多份简历后…出品 | OSC开源社区（ID：oschina2013)Sonatype 发布了最新的一份《软件供应链状况》报告，深入探讨了如何在充满选择的世界中定义更好的软件，并探讨人工智能 (AI) 对软件开发的深远
Java技术栈
0
springboot第70集：字节跳动后端三面经，一文让你走出微服务迷雾架构周刊
创建一个使用Kubernetes (K8s) 和 Jenkins 来自动化 GitLab 前端项目打包的CI/CD流水线，需要配置多个组件。下面，我将概述一个基本的设置步骤和示例脚本，以帮助你理解如何使用这些工具整合一个自动化流程。前提条件确保你已经有：Kubernetes 集群：用于部署 Jenk
程序源代码
0
1000Mbps换算成MB/s是多少？除以8？想简单了！
原文链接：https://post.smzdm.com/p/azoqenzp/在网络传输的时候，往往会用到Mbps这个单位，GbE or 1 GigE 的网卡现在很流行，这个东西被大家叫做“千兆网卡”。同时，大家特别习惯用GB或者MB来描述一个磁盘的大小。这个叫做Gigabyte或者Megabyte
测试开发技术
0
【第129期】程序员的新宠：三款终端工具，让你告别Xshell！
概述 WindTerm：跨平台的SSH利器 首先介绍的是WindTerm，这是一款使用C语言开发的跨平台SSH客户端。它不仅完全免费，而且没有商业使用的限制。WindTerm支持SSH v2、Telnet、Raw Tcp等协议，而且性能出色，甚至超过了FinalShell和Electerm。功能
前端微服务
0
Java与lua互相调用简单教程
来源：网络👉 欢迎加入小哈的星球 ，你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利全栈前后端分离博客项目 2.0 版本完结啦， 演示链接：http://116.62.199.48/ ，新项目
小哈学Java
0
美团社招一面，比预想的简单。
面试题大全：www.javacn.site面试这件事就很玄学，有时候你觉得他可能很难，但面完之后竟然出奇的顺利，问的问题你都会；有些你觉得这次面试应该很简单，但去了之后就被问懵了，所以面试这件事有很多一部分运气的成分。所以说，在没有 Offer 之前就是多准备、楞怂面，主打一个大力出奇迹。这不，逛牛
Java中文社群
0
点赞
评论
收藏
分享

手机扫一扫分享

举报