Flink 1.9 实战:使用 SQL 读取 Kafka 并写入 MySQL

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 《Flink SQL 1.9.0 技术内幕和最佳实践》,许多小伙伴对演示环节的 Demo 代码非常感兴趣,迫不及待地想尝试下,所以写了这篇文章分享下这份代码。

上周六在深圳分享了《Flink SQL 1.9.0 技术内幕和最佳实践》,会后许多小伙伴对最后演示环节的 Demo 代码非常感兴趣,迫不及待地想尝试下,所以写了这篇文章分享下这份代码。希望对于 Flink SQL 的初学者能有所帮助。完整分享可以观看 Meetup 视频回顾 :https://developer.aliyun.com/live/1416

演示代码已经开源到了 GitHub 上:https://github.com/wuchong/flink-sql-submit

这份代码主要由两部分组成:1) 能用来提交 SQL 文件的 SqlSubmit 实现。2) 用于演示的 SQL 示例、Kafka 启动停止脚本、 一份测试数据集、Kafka 数据源生成器。

通过本实战,你将学到:

  1. 如何使用 Blink Planner
  2. 一个简单的 SqlSubmit 是如何实现的
  3. 如何用 DDL 创建一个 Kafka 源表和 MySQL 结果表
  4. 运行一个从 Kafka 读取数据,计算 PVUV,并写入 MySQL 的作业
  5. 设置调优参数,观察对作业的影响

SqlSubmit 的实现

笔者一开始是想用 SQL Client 来贯穿整个演示环节,但可惜 1.9 版本 SQL CLI 还不支持处理 CREATE TABLE 语句。所以笔者就只好自己写了个简单的提交脚本。后来想想,也挺好的,可以让听众同时了解如何通过 SQL 的方式,和编程的方式使用 Flink SQL。

SqlSubmit 的主要任务是执行和提交一个 SQL 文件,实现非常简单,就是通过正则表达式匹配每个语句块。如果是 CREATE TABLE 或 INSERT INTO 开头,则会调用 tEnv.sqlUpdate(...)。如果是 SET 开头,则会将配置设置到 TableConfig 上。其核心代码主要如下所示:

EnvironmentSettings settings = EnvironmentSettings.newInstance()
        .useBlinkPlanner()
        .inStreamingMode()
        .build();
// 创建一个使用 Blink Planner 的 TableEnvironment, 并工作在流模式
TableEnvironment tEnv = TableEnvironment.create(settings);
// 读取 SQL 文件
List<String> sql = Files.readAllLines(path);
// 通过正则表达式匹配前缀,来区分不同的 SQL 语句
List<SqlCommandCall> calls = SqlCommandParser.parse(sql);
// 根据不同的 SQL 语句,调用 TableEnvironment 执行
for (SqlCommandCall call : calls) {
  switch (call.command) {
    case SET:
      String key = call.operands[0];
      String value = call.operands[1];
      // 设置参数
      tEnv.getConfig().getConfiguration().setString(key, value);
      break;
    case CREATE_TABLE:
      String ddl = call.operands[0];
      tEnv.sqlUpdate(ddl);
      break;
    case INSERT_INTO:
      String dml = call.operands[0];
      tEnv.sqlUpdate(dml);
      break;
    default:
      throw new RuntimeException("Unsupported command: " + call.command);
  }
}
// 提交作业
tEnv.execute("SQL Job");
AI 代码解读

使用 DDL 连接 Kafka 源表

在 flink-sql-submit 项目中,我们准备了一份测试数据集(来自阿里云天池公开数据集,特别鸣谢),位于 src/main/resources/user_behavior.log。数据以 JSON 格式编码,大概长这个样子:

{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
{"user_id": "662867", "item_id":"2244074", "category_id": "1575622", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
AI 代码解读

为了模拟真实的 Kafka 数据源,笔者还特地写了一个 source-generator.sh 脚本(感兴趣的可以看下源码),会自动读取 user_behavior.log 的数据并以默认每毫秒1条的速率灌到 Kafka 的 user_behavior topic 中。

有了数据源后,我们就可以用 DDL 去创建并连接这个 Kafka 中的 topic(详见 src/main/resources/q1.sql)。

CREATE TABLE user_log (
    user_id VARCHAR,
    item_id VARCHAR,
    category_id VARCHAR,
    behavior VARCHAR,
    ts TIMESTAMP
) WITH (
    'connector.type' = 'kafka', -- 使用 kafka connector
    'connector.version' = 'universal',  -- kafka 版本,universal 支持 0.11 以上的版本
    'connector.topic' = 'user_behavior',  -- kafka topic
    'connector.startup-mode' = 'earliest-offset', -- 从起始 offset 开始读取
    'connector.properties.0.key' = 'zookeeper.connect',  -- 连接信息
    'connector.properties.0.value' = 'localhost:2181', 
    'connector.properties.1.key' = 'bootstrap.servers',
    'connector.properties.1.value' = 'localhost:9092', 
    'update-mode' = 'append',
    'format.type' = 'json',  -- 数据源格式为 json
    'format.derive-schema' = 'true' -- 从 DDL schema 确定 json 解析规则
)
AI 代码解读

注:可能有用户会觉得其中的 connector.properties.0.key 等参数比较奇怪,社区计划将在下一个版本中改进并简化 connector 的参数配置。

使用 DDL 连接 MySQL 结果表

连接 MySQL 可以使用 Flink 提供的 JDBC connector。例如

CREATE TABLE pvuv_sink (
    dt VARCHAR,
    pv BIGINT,
    uv BIGINT
) WITH (
    'connector.type' = 'jdbc', -- 使用 jdbc connector
    'connector.url' = 'jdbc:mysql://localhost:3306/flink-test', -- jdbc url
    'connector.table' = 'pvuv_sink', -- 表名
    'connector.username' = 'root', -- 用户名
    'connector.password' = '123456', -- 密码
    'connector.write.flush.max-rows' = '1' -- 默认5000条,为了演示改为1条
)
AI 代码解读

PV UV 计算

假设我们的需求是计算每小时全网的用户访问量,和独立用户数。很多用户可能会想到使用滚动窗口来计算。但这里我们介绍另一种方式。即 Group Aggregation 的方式。

INSERT INTO pvuv_sink
SELECT
  DATE_FORMAT(ts, 'yyyy-MM-dd HH:00') dt,
  COUNT(*) AS pv,
  COUNT(DISTINCT user_id) AS uv
FROM user_log
GROUP BY DATE_FORMAT(ts, 'yyyy-MM-dd HH:00')
AI 代码解读

它使用 DATE_FORMAT 这个内置函数,将日志时间归一化成“年月日小时”的字符串格式,并根据这个字符串进行分组,即根据每小时分组,然后通过 COUNT(*) 计算用户访问量(PV),通过 COUNT(DISTINCT user_id) 计算独立用户数(UV)。这种方式的执行模式是每收到一条数据,便会进行基于之前计算的值做增量计算(如+1),然后将最新结果输出。所以实时性很高,但输出量也大。

我们将这个查询的结果,通过 INSERT INTO 语句,写到了之前定义的 pvuv_sink MySQL 表中。

注:在深圳 Meetup 中,我们有对这种查询的性能调优做了深度的介绍。

实战演示

环境准备

本实战演示环节需要安装一些必须的服务,包括:

  • Flink 本地集群:用来运行 Flink SQL 任务。
  • Kafka 本地集群:用来作为数据源。
  • MySQL 数据库:用来作为结果表。
  • Flink 本地集群安装

1.下载 Flink 1.9.0 安装包并解压
2.下载以下依赖 jar 包,并拷贝到 flink-1.9.0/lib/ 目录下。因为我们运行时需要依赖各个 connector 实现。

3.将 flink-1.9.0/conf/flink-conf.yaml 中的 taskmanager.numberOfTaskSlots 修改成 10,因为我们的演示任务可能会消耗多于1个的 slot。
4.在 flink-1.9.0 目录下执行 ./bin/start-cluster.sh,启动集群。

image.png

另外,还需要将 Flink 的安装路径填到 flink-sql-submit 项目的 env.sh 中,用于后面提交 SQL 任务,如我的路径是

FLINK_DIR=/Users/wuchong/dev/install/flink-1.9.0
AI 代码解读

Kafka 本地集群安装

将安装路径填到 flink-sql-submit 项目的 env.sh 中,如我的路径是

KAFKA_DIR=/Users/wuchong/dev/install/kafka_2.11-2.2.0
AI 代码解读

在 flink-sql-submit 目录下运行 ./start-kafka.sh 启动 Kafka 集群。

在命令行执行 jps,如果看到 Kafka 进程和 QuorumPeerMain 进程即表明启动成功。

MySQL 安装

可以在官方页面下载 MySQL 并安装:
https://dev.mysql.com/downloads/mysql/
如果有 Docker 环境的话,也可以直接通过 Docker 安装


然后在 MySQL 中创建一个 `flink-test` 的数据库,并按照上文的 schema 创建 `pvuv_sink` 表。

### 提交 SQL 任务

1.在 `flink-sql-submit` 目录下运行 `./source-generator.sh`,会自动创建 `user_behavior topic`,并实时往里灌入数据。

![image.png](https://ucc.alicdn.com/pic/developer-ecology/c914d05032e54bcc8c710f8327fc9e29.png)


2.在 `flink-sql-submit` 目录下运行 `./run.sh q1`, 提交成功后,可以在 Web UI 中看到拓扑。

![image.png](https://ucc.alicdn.com/pic/developer-ecology/1190a48a1fa140e78ae4d19811d23292.png)


在 MySQL 客户端,我们也可以实时地看到每个小时的 pv uv 值在不断地变化

![image.png](https://ucc.alicdn.com/pic/developer-ecology/22ce6c49dbb44aed97f1ff8e74adf474.png)


## 结尾

本文带大家搭建基础集群环境,并使用 SqlSubmit 提交纯 SQL 任务来学习了解如何连接外部系统。`flink-sql-submit/src/main/resources/q1.sql` 中还有一些注释掉的调优参数,感兴趣的同学可以将参数打开,观察对作业的影响。关于这些调优参数的原理,可以看下我在 [深圳 Meetup](https://developer.aliyun.com/article/716997?spm=a2c6h.12873581.0.0.795810c490Bj7l&groupCode=sc) 上的分享《Flink SQL 1.9.0 技术内幕和最佳实践》。

----

**▼ Apache Flink 社区推荐 ▼ **

Apache Flink 及大数据领域顶级盛会 Flink Forward Asia 2019 重磅开启,目前正在征集议题,限量早鸟票优惠ing。了解 Flink Forward Asia 2019 的更多信息,请查看:

https://developer.aliyun.com/special/ffa2019

首届 Apache Flink 极客挑战赛重磅开启,聚焦机器学习与性能优化两大热门领域,40万奖金等你拿,加入挑战请点击:
AI 代码解读
目录
打赏
0
2
0
2
36029
分享
相关文章
MySQL进阶突击系列(07) 她气鼓鼓递来一条SQL | 怎么看执行计划、SQL怎么优化?
在日常研发工作当中,系统性能优化,从大的方面来看主要涉及基础平台优化、业务系统性能优化、数据库优化。面对数据库优化,除了DBA在集群性能、服务器调优需要投入精力,我们研发需要负责业务SQL执行优化。当业务数据量达到一定规模后,SQL执行效率可能就会出现瓶颈,影响系统业务响应。掌握如何判断SQL执行慢、以及如何分析SQL执行计划、优化SQL的技能,在工作中解决SQL性能问题显得非常关键。
MySQL原理简介—1.SQL的执行流程
本文介绍了MySQL驱动、数据库连接池及SQL执行流程的关键组件和作用。主要内容包括:MySQL驱动用于建立Java系统与数据库的网络连接;数据库连接池提高多线程并发访问效率;MySQL中的连接池维护多个数据库连接并进行权限验证;网络连接由线程处理,监听请求并读取数据;SQL接口负责执行SQL语句;查询解析器将SQL语句解析为可执行逻辑;查询优化器选择最优查询路径;存储引擎接口负责实际的数据操作;执行器根据优化后的执行计划调用存储引擎接口完成SQL语句的执行。整个流程确保了高效、安全地处理SQL请求。
116 75
SpringBoot 通过集成 Flink CDC 来实时追踪 MySql 数据变动
通过详细的步骤和示例代码,您可以在 SpringBoot 项目中成功集成 Flink CDC,并实时追踪 MySQL 数据库的变动。
91 43
MySQL原理简介—10.SQL语句和执行计划
本文介绍了MySQL执行计划的相关概念及其优化方法。首先解释了什么是执行计划,它是SQL语句在查询时如何检索、筛选和排序数据的过程。接着详细描述了执行计划中常见的访问类型,如const、ref、range、index和all等,并分析了它们的性能特点。文中还探讨了多表关联查询的原理及优化策略,包括驱动表和被驱动表的选择。此外,文章讨论了全表扫描和索引的成本计算方法,以及MySQL如何通过成本估算选择最优执行计划。最后,介绍了explain命令的各个参数含义,帮助理解查询优化器的工作机制。通过这些内容,读者可以更好地理解和优化SQL查询性能。
【MySQL基础篇】全面学习总结SQL语法、DataGrip安装教程
本文详细介绍了MySQL中的SQL语法,包括数据定义(DDL)、数据操作(DML)、数据查询(DQL)和数据控制(DCL)四个主要部分。内容涵盖了创建、修改和删除数据库、表以及表字段的操作,以及通过图形化工具DataGrip进行数据库管理和查询。此外,还讲解了数据的增、删、改、查操作,以及查询语句的条件、聚合函数、分组、排序和分页等知识点。
【MySQL基础篇】全面学习总结SQL语法、DataGrip安装教程
Flink CDC MySQL同步MySQL错误记录
在使用Flink CDC同步MySQL数据时,常见的错误包括连接错误、权限错误、表结构变化、数据类型不匹配、主键冲突和
175 17
MySQL进阶突击系列(02)一条更新SQL执行过程 | 讲透undoLog、redoLog、binLog日志三宝
本文详细介绍了MySQL中update SQL执行过程涉及的undoLog、redoLog和binLog三种日志的作用及其工作原理,包括它们如何确保数据的一致性和完整性,以及在事务提交过程中各自的角色。同时,文章还探讨了这些日志在故障恢复中的重要性,强调了合理配置相关参数对于提高系统稳定性的必要性。
MySQL 高级(进阶) SQL 语句
MySQL 提供了丰富的高级 SQL 语句功能,能够处理复杂的数据查询和管理需求。通过掌握窗口函数、子查询、联合查询、复杂连接操作和事务处理等高级技术,能够大幅提升数据库操作的效率和灵活性。在实际应用中,合理使用这些高级功能,可以更高效地管理和查询数据,满足多样化的业务需求。
286 3
MySQL导入.sql文件后数据库乱码问题
本文分析了导入.sql文件后数据库备注出现乱码的原因,包括字符集不匹配、备注内容编码问题及MySQL版本或配置问题,并提供了详细的解决步骤,如检查和统一字符集设置、修改客户端连接方式、检查MySQL配置等,确保导入过程顺利。
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")

相关产品

  • 实时计算 Flink版