Flink使用JdbcSink下沉数据到数据库MySQL

Flink官网提供了JdbcSink的功能，如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.fromElements(...).addSink(JdbcSink.sink("insert into books (id, title, author, price, qty) values (?,?,?,?,?)",(ps, t) -> {ps.setInt(1, t.id);ps.setString(2, t.title);ps.setString(3, t.author);ps.setDouble(4, t.price);ps.setInt(5, t.qty);},new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl(getDbMetadata().getUrl()).withDriverName(getDbMetadata().getDriverClass()).build()));
env.execute();

但是，在下才疏学浅，中间那个…看的在下一脸懵逼。
自己从网上查找资料发现只有创建一个RichSinkFunction的方法，在下就是想用官网的那个方法。于是经过在gitee寻找依赖，在代码中寻找关键字JdbcSink，终于找到了，Apache Flink的整个项目的代码。起初我还不知道，发现这个demo项目怎么这么大，还有几个依赖下载不下来。
在Flink的测试类中找到了JdbcSink相关的Java测试代码。
然后经过在下的不懈努力，终于写成了一个scala的测试类。如下：

import org.apache.flink.connector.jdbc.JdbcConnectionOptions.JdbcConnectionOptionsBuilder
import org.apache.flink.connector.jdbc.{JdbcSink, JdbcStatementBuilder}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}import java.sql.{PreparedStatement, Types}case class TestEntry(id: Int, title: String, author: String, price: Double, qty: Int)object JDBCSinkTestJob {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)env.fromElements(TestEntry(1001, "Java public for dummies", "Tan Ah Teck", 11.11, 11),TestEntry(1002, "More Java for dummies", "Tan Ah Teck", 22.22, 22),TestEntry(1003, "More Java for more dummies", "Mohammad Ali", 33.33, 33),TestEntry(1004, "A Cup of Java", "Kumar", 44.44, 44),TestEntry(1005, "A Teaspoon of Java", "Kevin Jones", 55.55, 55),TestEntry(1006, "A Teaspoon of Java 1.4", "Kevin Jones", 66.66, 66),TestEntry(1007, "A Teaspoon of Java 1.5", "Kevin Jones", 77.77, 77),TestEntry(1008, "A Teaspoon of Java 1.6", "Kevin Jones", 88.88, 88),TestEntry(1009, ("A Teaspoon of Java 1.7"), ("Kevin Jones"), 99.99, 99),TestEntry(1010, ("A Teaspoon of Java 1.8"), ("Kevin Jones"), 99, 1010)).addSink(JdbcSink.sink(String.format("insert into %s (id, title, author, price, qty) values (?,?,?,?,?)", "books"),new JdbcStatementBuilder[TestEntry] {override def accept(ps: PreparedStatement, t: TestEntry): Unit = {ps.setInt(1, t.id);ps.setString(2, t.title);ps.setString(3, t.author);if (t.price == null) {ps.setNull(4, Types.DOUBLE);} else {ps.setDouble(4, t.price);}ps.setInt(5, t.qty);}},new JdbcConnectionOptionsBuilder().withUrl("jdbc:mysql://hadoop10:3306/test?useUnicode=true&characterEncoding=UTF-8").withUsername("root").withPassword("123456").withDriverName("com.mysql.jdbc.Driver").build()))env.execute()}}

这个测试类，把测试代码的变量给提取了出来。
写代码的过程还需要导入MySQL需要的依赖，自行搜索。
测试结果

Perfect，在下非常的高兴，决定打两把游戏奖赏一下自己。

参考资料：
gitee中的Flink项目代码

Flink使用JdbcSink下沉数据到数据库MySQL相关推荐

大数据之数据库mysql优化实战（一）
2019独角兽企业重金招聘Python工程师标准>>> :facepunch: 大数据之数据库mysql优化实战(一) 首先你要有数据,不然怎么测试,几百条就算了,还没跑就完了. 本 ...
mysql查出倒序第一条数据_[数据库]mysql 记录根据日期字段倒序输出
[数据库]mysql 记录根据日期字段倒序输出 0 2016-07-21 11:00:17 我们知道倒序输出是很简单的 select * from table order by id desc 直接这 ...
python网页数据存入数据库_python网络爬虫抓取动态网页并将数据存入数据库MySQL...
简述以下的代码是使用python实现的网络爬虫,抓取动态网页 http://hb.qq.com/baoliao/ .此网页中的最新.精华下面的内容是由JavaScript动态生成的.审查网页元素与网 ...
python网站数据写入mysql_python网络爬虫抓取动态网页并将数据存入数据库MySQL
简述以下的代码是使用python实现的网络爬虫,抓取动态网页 http://hb.qq.com/baoliao/ .此网页中的最新.精华下面的内容是由JavaScript动态生成的.审查网页元素与网 ...
Java 批量插入数据到数据库(MySQL)中
实现Java批量插入数据库数据: package Proxy;import java.io.BufferedReader; import java.io.File; import java.io.Fi ...
excel表数据导入数据库mysql中，并解决导入时间格式问题
1.准备好Excel表数据 id category_id category_pid title art_desc content imageurl tags create_time 3 1 E ...
mysql 导出中间数据_MYSQL数据库之间的数据导出与导入
源数据库地址: 172.16.1.7 目标数据库地址: 172.16.1.51 步骤: (1) 进入172.16.1.7服务器,登录mysql数据库 mysqldump -uusername -ppa ...
nodejs操作sqlserver数据_pyspark操作MySQL、SQLServer数据库进行数据处理操作
欢迎访问本人的CSDN博客[Together_CZ],我是沂水寒城. https://yishuihancheng.blog.csdn.net 在大数据处理领域里面,Hadoop和spark可以说是最 ...
python爬取新闻并归数据库_Python爬取数据并写入MySQL数据库操作示例
Python爬取数据并写入MySQL数据库的实例首先我们来爬取 http://html-color-codes.info/color-names/ 的一些数据. 按 F12 或 ctrl+u 审查元 ...

Flink使用JdbcSink下沉数据到数据库MySQL

Flink使用JdbcSink下沉数据到数据库MySQL相关推荐

最新文章

热门文章