17-Flink消费Kafka写入Mysql
时间:2022-06-18
本文章向大家介绍17-Flink消费Kafka写入Mysql,主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。
戳更多文章:
...
本文介绍消费Kafka的消息实时写入Mysql。
- maven新增依赖:
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.39</version>
</dependency>
2.重写RichSinkFunction,实现一个Mysql Sink
public class MysqlSink extends
RichSinkFunction<Tuple3<Integer, String, Integer>> {
private Connection connection;
private PreparedStatement preparedStatement;
String username = "";
String password = "";
String drivername = ""; //配置改成自己的配置
String dburl = "";
@Override
public void invoke(Tuple3<Integer, String, Integer> value) throws Exception {
Class.forName(drivername);
connection = DriverManager.getConnection(dburl, username, password);
String sql = "replace into table(id,num,price) values(?,?,?)"; //假设mysql 有3列 id,num,price
preparedStatement = connection.prepareStatement(sql);
preparedStatement.setInt(1, value.f0);
preparedStatement.setString(2, value.f1);
preparedStatement.setInt(3, value.f2);
preparedStatement.executeUpdate();
if (preparedStatement != null) {
preparedStatement.close();
}
if (connection != null) {
connection.close();
}
}
}
- Flink主类
public class MysqlSinkTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
// 1,abc,100 类似这样的数据,当然也可以是很复杂的json数据,去做解析
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
env.getConfig().disableSysoutLogging(); //设置此可以屏蔽掉日记打印情况
env.getConfig().setRestartStrategy(
RestartStrategies.fixedDelayRestart(5, 5000));
env.enableCheckpointing(2000);
DataStream<String> stream = env
.addSource(consumer);
DataStream<Tuple3<Integer, String, Integer>> sourceStream = stream.filter((FilterFunction<String>) value -> StringUtils.isNotBlank(value))
.map((MapFunction<String, Tuple3<Integer, String, Integer>>) value -> {
String[] args1 = value.split(",");
return new Tuple3<Integer, String, Integer>(Integer
.valueOf(args1[0]), args1[1],Integer
.valueOf(args1[2]));
});
sourceStream.addSink(new MysqlSink());
env.execute("data to mysql start");
}
}
所有代码,我放在了我的公众号,回复Flink可以下载
- 海量【java和大数据的面试题+视频资料】整理在公众号,关注后可以下载~
- 更多大数据技术欢迎和作者一起探讨~
- Leetcode-Easy 543. Diameter of Binary Tree
- Leetcode-Easy 572. Subtree of Another Tree
- 图解javascript this指向什么?
- 2017/6/8-python正则表达式的使用
- 洛谷P1306 斐波那契公约数
- Angular开发实践(二):HRM运行机制
- Angular开发实践(一):环境准备及框架搭建
- 洛谷P2818 天使的起誓
- 连续子数组的最大和
- 10.25解题报告
- React第三方组件5(状态管理之Redux的使用①简单使用)
- JavaScript设计模式与开发实践 - 观察者模式
- React第三方组件4(状态管理之Reflux的使用⑤异步操作)
- Leetcode-Easy 804. Unique Morse Code Words
- MySQL 教程
- MySQL 安装
- MySQL 管理与配置
- MySQL PHP 语法
- MySQL 连接
- MySQL 创建数据库
- MySQL 删除数据库
- MySQL 选择数据库
- MySQL 数据类型
- MySQL 创建数据表
- MySQL 删除数据表
- MySQL 插入数据
- MySQL 查询数据
- MySQL where 子句
- MySQL UPDATE 查询
- MySQL DELETE 语句
- MySQL LIKE 子句
- mysql order by
- Mysql Join的使用
- MySQL NULL 值处理
- MySQL 正则表达式
- MySQL 事务
- MySQL ALTER命令
- MySQL 索引
- MySQL 临时表
- MySQL 复制表
- 查看MySQL 元数据
- MySQL 序列 AUTO_INCREMENT
- MySQL 处理重复数据
- MySQL 及 SQL 注入
- MySQL 导出数据
- MySQL 导入数据
- MYSQL 函数大全
- MySQL Group By 实例讲解
- MySQL Max()函数实例讲解
- mysql count函数实例
- MYSQL UNION和UNION ALL实例
- MySQL IN 用法
- MySQL between and 实例讲解