为什么Flink无法实时写入MySQL

发布时间：2021-08-25 21:07:29 来源：亿速云阅读：174 作者：chen 栏目：大数据

这篇文章主要介绍“为什么Flink无法实时写入MySQL”，在日常操作中，相信很多人在为什么Flink无法实时写入MySQL问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”为什么Flink无法实时写入MySQL”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

本文主要分为以下三部分：

问题描述
解决思路
原因剖析

问题描述

Flink 1.10 使用 flink-jdbc 连接器的方式与 MySQL 交互，读数据和写数据都能完成，但是在写数据时，发现 Flink 程序执行完毕之后，才能在 MySQL 中查询到插入的数据。即，虽然是流计算，但却不能实时的输出计算结果？

相关代码片段：

JDBCAppendTableSink.builder()    .setDrivername("com.mysql.jdbc.Driver")    .setDBUrl("jdbc:mysql://localhost/flink")    .setUsername("root")    .setPassword("123456")    .setParameterTypes(    BasicTypeInfo.INT_TYPE_INFO,    BasicTypeInfo.STRING_TYPE_INFO)    .setQuery("insert into batch_size values(?,?)")   .build()

如何解决？

Flink 1.10 这个问题是知道一秒钟，不知磨洋工的 Case，在初学时候非常容易遇上，那么真的是 Flink 不能实时写入 MySQL 吗？当然不是，上面代码基础之上简单的加上一行，就解决问题了：

....setBatchSize(1) //将写入MySQL的buffer大小为1。..

原因剖析

那么问题虽然解决了，根本原因是个啥呢？也许你看到这里会说，这问题很明显，就是 Flink 设计 JDBC Sink 的时候出于性能因素考虑，对写入 buffer 做了默认值设置。

没错，这一点你说的很对，在 Flink 1.10 中 JDBC OutputFormat 的基类 AbstractJDBCOutputFormat 里面和这相关的变量 DEFAULT_FLUSH_MAX_SIZE 默认值是 5000，所以在你学习测试时候由于测试数据少（少于 5000），数据一直在 buffer 中，直到数据源数据结束，作业也结束了，才将计算结果刷入 MySQL，所以没有实时的（每条）写入 MySQL。如下：

但这里还有个因素需要注意，那就是时间因素，上面 DEFAULT_FLUSH_INTERVAL_MILLS 默认值是 0，这个相当于没有时间限制，一直等到 buffer 满了或者作业结束才能触发写出动作。

也就是有些初学者，发现问题，即使故意 debug 时候打上断点，不让作业结束，但是等到花儿都谢了，数据也没有写入到 MySQL。

在 Flink 1.10 中 AbstractJDBCOutputFormat 有两个实现类：

分别对应了如下两类 Sink：

所以在 Flink 1.10 中不论是 AppendTableSink 和 UpsertTableSink 都会有同样的问题。不过 UpsertTableSink 时用户可以设置时间，而 AppendTableSink 是连时间设置的入口都木有。

到此，关于“为什么Flink无法实时写入MySQL”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注亿速云网站，小编会继续努力为大家带来更多实用的文章！

向AI问一下细节

为什么Flink无法实时写入MySQL

猜你喜欢

最新资讯

相关推荐

相关标签