大数据-玩转数据-Flink 自定义Sink(Mysql)

一、说明

如果Flink没有提供给我们可以直接使用的连接器,那我们如果想将数据存储到我们自己的存储设备中,mysql 的安装使用请参考
mysql-玩转数据-centos7下mysql的安装
创建表

CREATE TABLE `sensor` (`id` int(10)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

二、pom.xml 导入驱动

<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.33</version>
</dependency>

三、编写程序

package com.lyh.flink06;import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;public class SinkMysql {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);DataStreamSource<Integer> dataStreamSource = env.fromElements(1, 2, 3, 4, 5, 6);KeyedStream<Integer, Integer> keyedStream = dataStreamSource.keyBy(new KeySelector<Integer, Integer>() {@Overridepublic Integer getKey(Integer value) throws Exception {return value.intValue();}});keyedStream.addSink(new MysqlSink());env.execute();}public static class  MysqlSink extends RichSinkFunction<Integer>{private Connection sunbo;@Overridepublic void open(Configuration parameters) throws Exception {Class.forName("com.mysql.cj.jdbc.Driver");sunbo = DriverManager.getConnection("jdbc:mysql://192.168.220.100:3306/test?useSSL=false", "sunbo", "Mysql123456#");}@Overridepublic void close() throws Exception {if (sunbo != null) {sunbo.close();}}@Overridepublic void invoke(Integer value, Context context) throws Exception {String sql = "insert into sensor(id)values(?)";PreparedStatement ps = sunbo.prepareStatement(sql);ps.setInt(1,value.intValue());ps.execute();ps.close();}}
}

四、运行测试

在这里插入图片描述

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

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

相关文章

二 根据用户行为数据创建ALS模型并召回商品

二 根据用户行为数据创建ALS模型并召回商品 2.0 用户行为数据拆分 方便练习可以对数据做拆分处理 pandas的数据分批读取 chunk 厚厚的一块 相当大的数量或部分 import pandas as pd reader pd.read_csv(behavior_log.csv,chunksize100,iteratorTrue) count 0; for chunk in …

DNS协议及其工作原理

DNS是域名系统&#xff08;Domain Name System&#xff09;的缩写&#xff0c;它是一种用于将域名转换为IP地址的分布式数据库系统。它是因特网的基石&#xff0c;能够使人们通过域名方便地访问互联网&#xff0c;而无需记住复杂的IP地址。 DNS的历史可以追溯到1983年&#xf…

4个简化IT服务台任务的ChatGPT功能

最近几个月&#xff0c;ChatGPT 风靡全球&#xff0c;这是一个 AI 聊天机器人&#xff0c;使用户能够生成脚本、文章、锻炼图表等。这项技术在各行各业都有无穷无尽的应用&#xff0c;在本文中&#xff0c;我们将研究这种现代技术如何帮助服务台团队增强服务交付和客户体验。 什…

最佳实践:如何优雅地提交一个 Amazon EMR Serverless 作业?

《大数据平台架构与原型实现&#xff1a;数据中台建设实战》一书由博主历时三年精心创作&#xff0c;现已通过知名IT图书品牌电子工业出版社博文视点出版发行&#xff0c;点击《重磅推荐&#xff1a;建大数据平台太难了&#xff01;给我发个工程原型吧&#xff01;》了解图书详…

章节7:XSS检测和利用

章节7&#xff1a;XSS检测和利用 测试payload <script>alert(XSS)</script> <script>alert(document.cookie)</script> ><script>alert(document.cookie)</script> ><script>alert(document.cookie)</script> &qu…

元宇宙之经济(02)理解NFT

1 NFT是什么&#xff1f; 想象一下&#xff0c;你小时候曾经在操场上集齐过各种不同的贴纸&#xff0c;然后和朋友们交换&#xff0c;这些贴纸有着独特的图案和价值。NFT的概念与此类似&#xff0c;但在数字世界中运作。NFT是一种基于区块链技术的数字资产&#xff0c;每个NFT…

golang—面试题大全

目录标题 sliceslice和array的区别slice扩容机制slice是否线程安全slice分配到栈上还是堆上扩容过程中是否重新写入go深拷贝发生在什么情况下&#xff1f;切片的深拷贝是怎么做的copy和左值进行初始化区别slice和map的区别 mapmap介绍map的key的类型map对象如何比较map的底层原…

《Java极简设计模式》第03章:工厂方法模式(FactoryMethod)

作者&#xff1a;冰河 星球&#xff1a;http://m6z.cn/6aeFbs 博客&#xff1a;https://binghe.gitcode.host 文章汇总&#xff1a;https://binghe.gitcode.host/md/all/all.html 源码地址&#xff1a;https://github.com/binghe001/java-simple-design-patterns/tree/master/j…

无法正确识别车牌(Python、OpenCv、Tesseract)

我正在尝试识别车牌&#xff0c;但出现了错误&#xff0c;例如错误/未读取字符 以下是每个步骤的可视化&#xff1a; 从颜色阈值变形关闭获得遮罩 以绿色突出显示的车牌轮廓过滤器 将板轮廓粘贴到空白遮罩上 Tesseract OCR的预期结果 BP 1309 GD 但我得到的结果是 BP 1309…

腾讯云标准型CVM云服务器详细介绍

腾讯云CVM服务器标准型实例的各项性能参数平衡&#xff0c;标准型云服务器适用于大多数常规业务&#xff0c;例如&#xff1a;web网站及中间件等&#xff0c;常见的标准型云服务器有CVM标准型S5、S6、SA3、SR1、S5se等规格&#xff0c;腾讯云服务器网来详细说下云服务器CVM标准…

NAS搭建指南一——服务器的选择与搭建

一、服务器的选择 有自己的本地的公网 IP 的请跳过此篇文章按需求选择一个云服务器&#xff0c;目的就是为了进行 frp 的搭建&#xff0c;完成内网穿透我选择的是腾讯云服务器&#xff0c;我的配置如下&#xff0c;仅供参考&#xff1a; 4. 腾讯云服务器官网地址 二、服务器…

docker 镜像的导出与导入 save 与 load

一、镜像导出 docker save 导出 将系统中的镜像保存为压缩包&#xff0c;进行文件传输。使用 docker save --help 查看命令各参数&#xff0c;或者去docker官网查看.以 hello-world镜像为例。 A&#xff1a;将镜像保存为tar包 docker save image > package.tar docker sa…

day9 10-牛客67道剑指offer-JZ66、19、20、75、23、76、8、28、77、78

文章目录 1. JZ66 构建乘积数组暴力解法双向遍历 2. JZ19 正则表达式匹配3. JZ20 表示数值的字符串有限状态机遍历 4. JZ75 字符流中第一个不重复的字符5. JZ23 链表中环的入口结点快慢指针哈希表 6. JZ76 删除链表中重复的结点快慢指针三指针如果只保留一个重复结点 7. JZ8 二…

gitblit-使用

1.登入GitBlit服务器 默认用户和密码: admin/admin 2.创建一个新的版本库 点击图中的“版本库”&#xff0c;然后点击图中“创建版本库” 填写名称和描述&#xff0c;注意名称最后一定要加 .git选择限制查看、克隆和推送勾选“加入README”和“加入.gitignore文件”在图中的1处…

使用IIS服务器部署Flask python Web项目

参考文章 ""D:\Program Files (x86)\Python310\python310.exe"|"D:\Program Files (x86)\Python310\lib\site-packages\wfastcgi.py"" can now be used as a FastCGI script processor参考文章 请求路径填写*&#xff0c;模块选择FastCgiModule&…

一键部署 Umami 统计个人网站访问数据

谈到网站统计&#xff0c;大家第一时间想到的肯定是 Google Analytics。然而&#xff0c;我们都知道 Google Analytics 会收集所有用户的信息&#xff0c;对数据没有任何控制和隐私保护。 Google Analytics 收集的指标实在是太多了&#xff0c;有很多都是不必要的&#xff0c;…

Javascript 深入了解map

map() 是 JavaScript 数组提供的一个高阶函数&#xff0c;它用于对数组中的每个元素执行指定的函数&#xff0c;并返回一个新的数组&#xff0c;新数组中的元素是原数组中的每个元素经过函数处理后的结果。 map() 函数的语法如下&#xff1a; javascript array.map(callback(…

Multi-object navigation in real environments using hybrid policies 论文阅读

论文信息 题目&#xff1a;Multi-object navigation in real environments using hybrid policies 作者&#xff1a;Assem Sadek, Guillaume Bono 来源&#xff1a;CVPR 时间&#xff1a;2023 Abstract 机器人技术中的导航问题通常是通过 SLAM 和规划的结合来解决的。 最近…

优化堆排序(Java 实例代码)

目录 优化堆排序 Java 实例代码 src/runoob/heap/HeapSort.java 文件代码&#xff1a; 优化堆排序 上一节的堆排序&#xff0c;我们开辟了额外的空间进行构造堆和对堆进行排序。这一小节&#xff0c;我们进行优化&#xff0c;使用原地堆排序。 对于一个最大堆&#xff0c;首…

【设计模式】-策略模式:优雅处理条件逻辑

Java 策略模式之优雅处理条件逻辑 前言 在软件开发中&#xff0c;我们经常会遇到根据不同的条件执行不同逻辑的情况。这时&#xff0c;策略模式是一种常用的设计模式&#xff0c;能够使代码结构清晰、易于扩展和维护。 本文将详细介绍策略模式的概念及其在Java中的应用&#x…