Flink将数据写入MySQL(JDBC)

一、写在前面

在实际的生产环境中,我们经常会把Flink处理的数据写入MySQL、Doris等数据库中,下面以MySQL为例,使用JDBC的方式将Flink的数据实时数据写入MySQL。

二、代码示例

2.1 版本说明

        <flink.version>1.14.6</flink.version><spark.version>2.4.3</spark.version><hadoop.version>2.8.5</hadoop.version><hbase.version>1.4.9</hbase.version><hive.version>2.3.5</hive.version><java.version>1.8</java.version><scala.version>2.11.8</scala.version><mysql.version>8.0.22</mysql.version><scala.binary.version>2.11</scala.binary.version>

2.2 导入相关依赖

 <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_2.11</artifactId><version>${flink.version}</version>
</dependency>
<!--mysql连接器依赖-->
<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.22</version>
</dependency>

2.3 连接数据库,创建表

mysql> CREATE TABLE `ws` ( `id` varchar(100) NOT NULL,`ts` bigint(20) DEFAULT NULL,`vc` int(11) DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8

2.4 创建POJO类

package com.flink.POJOs;import java.util.Objects;/*** TODO POJO类的特点* 类是公有(public)的* 有一个无参的构造方法* 所有属性都是公有(public)的* 所有属性的类型都是可以序列化的*/
public class WaterSensor {//类的公共属性public String id;public Long ts;public Integer vc;//无参构造方法public WaterSensor() {//System.out.println("调用了无参数的构造方法");}public WaterSensor(String id, Long ts, Integer vc) {this.id = id;this.ts = ts;this.vc = vc;}//生成get和set方法public void setId(String id) {this.id = id;}public void setTs(Long ts) {this.ts = ts;}public void setVc(Integer vc) {this.vc = vc;}public String getId() {return id;}public Long getTs() {return ts;}public Integer getVc() {return vc;}//重写toString方法@Overridepublic String toString() {return "WaterSensor{" +"id='" + id + '\'' +", ts=" + ts +", vc=" + vc +'}';}//重写equals和hasCode方法@Overridepublic boolean equals(Object o) {if (this == o) return true;if (o == null || getClass() != o.getClass()) return false;WaterSensor that = (WaterSensor) o;return id.equals(that.id) && ts.equals(that.ts) && vc.equals(that.vc);}@Overridepublic int hashCode() {return Objects.hash(id, ts, vc);}
}
//scala的case类?

2.5 自定义map函数

package com.flink.POJOs;import org.apache.flink.api.common.functions.MapFunction;public class WaterSensorMapFunction implements MapFunction<String, WaterSensor> {@Overridepublic WaterSensor map(String value) throws Exception {String[] datas = value.split(",");return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));}
}

2.5 Flink2MySQL

package com.flink.DataStream.Sink;import com.flink.POJOs.WaterSensor;
import com.flink.POJOs.WaterSensorMapFunction;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;import java.sql.PreparedStatement;
import java.sql.SQLException;/*** Flink 输出到 MySQL(JDBC)*/
public class flinkSinkJdbc {public static void main(String[] args) throws Exception {//TODO 创建Flink上下文执行环境StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();streamExecutionEnvironment.setParallelism(1);//TODO SourceDataStreamSource<String> dataStreamSource = streamExecutionEnvironment.socketTextStream("localhost", 8888);//TODO TransferSingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = dataStreamSource.map(new WaterSensorMapFunction());/**TODO 写入 mysql* 1、只能用老的 sink 写法* 2、JDBCSink 的 4 个参数:*   第一个参数: 执行的 sql,一般就是 insert into*   第二个参数: 预编译 sql, 对占位符填充值*   第三个参数: 执行选项 ---->攒批、重试*   第四个参数: 连接选项---->url、用户名、密码*/SinkFunction<WaterSensor> sinkFunction = JdbcSink.sink("insert into ws values(?,?,?)",new JdbcStatementBuilder<WaterSensor>() {@Overridepublic void accept(PreparedStatement preparedStatement, WaterSensor waterSensor) throws SQLException {preparedStatement.setString(1, waterSensor.getId());preparedStatement.setLong(2, waterSensor.getTs());preparedStatement.setInt(3, waterSensor.getVc());System.out.println("数据写入成功:"+'('+waterSensor.getId()+","+waterSensor.getTs()+","+waterSensor.getVc()+")");}}, JdbcExecutionOptions.builder().withMaxRetries(3)         // 重试次数.withBatchSize(100)        // 批次的大小:条数.withBatchIntervalMs(3000) // 批次的时间.build(),new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl("jdbc:mysql://localhost:3306/dw?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8").withUsername("root").withPassword("********").withConnectionCheckTimeoutSeconds(60) // 重试的超时时间.build());//TODO 写入到MysqlwaterSensorSingleOutputStreamOperator.addSink(sinkFunction);streamExecutionEnvironment.execute();}
}

2.6 启动necat、Flink,观察数据库写入情况

nc -lk 9999 #启动necat、并监听8888端口,写入数据

在这里插入图片描述
启动Flink程序
在这里插入图片描述
查看数据库写入是否正常
在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/124625.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

故障诊断模型 | Maltab实现BiLSTM双向长短期记忆神经网络故障诊断

文章目录 效果一览文章概述模型描述源码设计参考资料效果一览 文章概述 故障诊断模型 | Maltab实现BiLSTM双向长短期记忆神经网络故障诊断 模型描述 利用各种检查和测试方法,发现系统和设备是否存在故障的过程是故障检测;而进一步确定故障所在大致部位的过程是故障定位。故障…

Linux网络编程二(TCP三次握手、四次挥手、TCP滑动窗口、MSS、TCP状态转换、多进程/多线程服务器实现)

TCP三次握手 TCP三次握手(TCP three-way handshake)是TCP协议建立可靠连接的过程&#xff0c;确保客户端和服务器之间可以进行可靠的通信。下面是TCP三次握手的详细过程&#xff1a; 假设客户端为A&#xff0c;服务器为B 1、第一次握手&#xff08;SYN1&#xff0c;seq500&…

03_Flutter自定义下拉菜单

03_Flutter自定义下拉菜单 在Flutter的内置api中&#xff0c;可以使用showMenu实现类似下拉菜单的效果&#xff0c;或者使用PopupMenuButton组件&#xff0c;PopupMenuButton内部也是使用了showMenu这个api&#xff0c;但是使用showMenu时&#xff0c;下拉面板的显示已经被约定…

List的add(int index,E element)陷阱,不得不防

项目场景&#xff1a; 项目中有两个List列表&#xff0c;一个是List1用来存储一个标识&#xff0c;后续会根据这个标识去重。 一个List2是用来返回对象的&#xff0c;其中对象里也有一个属性List3。现需要将重复的标识数据追加到List3 我想到的两个方案&#xff1a; 尽量不动…

吴恩达《机器学习》2-5->2-7:梯度下降算法与理解

一、梯度下降算法 梯度下降算法的目标是通过反复迭代来更新模型参数&#xff0c;以便最小化代价函数。代价函数通常用于衡量模型的性能&#xff0c;我们希望找到使代价函数最小的参数值。这个过程通常分为以下几个步骤&#xff1a; 初始化参数&#xff1a; 随机或设定初始参数…

项目资源管理-考试重点

1. 冲突管理的5种方法 ① 撤退/回避 ② 缓和/包容 ③ 妥协/调解 ④ 强迫/命令 ⑤ 合作/解决问题 2. 虚拟团队的优缺点 优点&#xff1a; ① 能够利用不在同一地理区域的专家的专业技术 ② 将在家办公的员工纳入团队 ③ 以及将行动不便者或残疾人纳入团队 缺点&#…

【图像分类】基于计算机视觉的坑洼道路检测和识别(ResNet网络,附代码和数据集)

写在前面: 首先感谢兄弟们的关注和订阅,让我有创作的动力,在创作过程我会尽最大能力,保证作品的质量,如果有问题,可以私信我,让我们携手共进,共创辉煌。 本篇博文,我们将使用PyTorch深度学习框架搭建ResNet实现钢轨缺陷识别,附完整的项目代码和数据集,可以说是全网…

C++ 自引用指针this(整理)

使用例子&#xff1a; #include <iostream> #include <Windows.h> using namespace std; class A { public:A(int x1){x x1;}void disp(){cout<<"this"<<this<<" when x"<<this->x<<endl;} private:int x;…

Node学习笔记之user用户API模块

1、获取用户的基本信息 步骤 获取登录会话存储的session中用户的id判断是否获取到id根据用户id查询数据库中的个人信息检查指定 id 的用户是否存在将密码设置为空将数据返回给前端 // 获取用户信息数据 exports.userinfo (req, res) > {(async function () {// 1. 获取…

关于Spring和SpringBoot中对配置文件的读取

Spring读取xml文件 具体流程见网址Spring源码分析2 — spring XML配置文件的解析流程 - 知乎 (zhihu.com) 我这里只做一下总结和自己的理解&#xff1a; &#xff08;1&#xff09;通过getConfigLocations方法, 从web.xml文件中读取配置文件地址&#xff0c;如果web.xml中读取…

合肥中科深谷嵌入式项目实战——人工智能与机械臂(四)

订阅&#xff1a;新手可以订阅我的其他专栏。免费阶段订阅量1000 python项目实战 Python编程基础教程系列&#xff08;零基础小白搬砖逆袭) 作者&#xff1a;爱吃饼干的小白鼠。Python领域优质创作者&#xff0c;2022年度博客新星top100入围&#xff0c;荣获多家平台专家称号。…

对象补充-原型和函数原型-创建对象

defineProperties可以定义多个属性描述符 var obj {// 私有属性&#xff08;js里面是没有严格意义的私有属性&#xff09;_age: 18,_eat: function() {} }Object.defineProperties(obj, {name: {configurable: true,enumerable: true,writable: true,value: "why"}…

Failed to prepare the device for development

&#x1f468;&#x1f3fb;‍&#x1f4bb; 热爱摄影的程序员 &#x1f468;&#x1f3fb;‍&#x1f3a8; 喜欢编码的设计师 &#x1f9d5;&#x1f3fb; 擅长设计的剪辑师 &#x1f9d1;&#x1f3fb;‍&#x1f3eb; 一位高冷无情的编码爱好者 大家好&#xff0c;我是 DevO…

创新领航 | 竹云参编《基于区块链的数据资产评估实施指南》正式发布!

10月25日&#xff0c;由深圳数宝数据服务股份有限公司和深圳职业技术大学提出&#xff0c;中国科学院深圳先进技术研究院、中国电子技术标准化研究院、中国&#xff08;天津&#xff09;自由贸易试验区政策与产业创新发展局、网络空间治理与数字经济法治&#xff08;长三角&…

论文阅读——ELECTRA

论文下载&#xff1a;https://openreview.net/pdf?idr1xMH1BtvB 另一篇分析文章&#xff1a;ELECTRA 详解 - 知乎 一、概述 对BERT的token mask 做了改进。结合了GAN生成对抗模型的思路&#xff0c;但是和GAN不同。 不是对选择的token直接用mask替代&#xff0c;而是替换为…

喜讯!合合信息顺利通过CMMI3级评估

近日&#xff0c;在擎标顾问团的咨询辅导下&#xff0c;上海合合信息科技股份有限公司&#xff08;简称“合合信息”&#xff09;顺利通过了CMMI3级评估。CMMI是国际上最流行、最实用的一种软件生产过程标准和软件企业成熟度等级认证的标准&#xff0c;通过该认证表明企业在开发…

【多线程面试题十四】、说一说synchronized的底层实现原理

文章底部有个人公众号&#xff1a;热爱技术的小郑。主要分享开发知识、学习资料、毕业设计指导等。有兴趣的可以关注一下。为何分享&#xff1f; 踩过的坑没必要让别人在再踩&#xff0c;自己复盘也能加深记忆。利己利人、所谓双赢。 面试官&#xff1a;说一说synchronized的底…

常见面试题-MySQL专栏(二)

了解索引扫描吗&#xff1f; 答&#xff1a; MySQL有两种方法生成有序结果&#xff1a; 通过排序操作按照索引顺序扫描 如果 explain 出来的 type 列值为 “index” 的话&#xff0c;说明是按照索引扫描了。 索引扫描本身的速度是很快的。但是如果索引不能覆盖查询所需的全…

Visual Studio Code的下载与安装

Visual Studio Code&#xff08;简称 VS Code&#xff09;是由 Microsoft 开发的免费、开源的文本编辑器&#xff0c;适用于多种操作系统&#xff0c;包括 Windows、macOS 和 Linux。它的设计目标是成为一款轻量级、高效的代码编辑工具&#xff0c;同时提供丰富的扩展和功能&am…

微软服务器数据库 Navicat Premium 连接

需要固定IP&#xff0c;在服务器的网络里面加上。 需要打开SSL,入下图 只用打开&#xff0c;不用选择秘钥&#xff0c;证书等