flink 中自身虽然实现了大量的connectors，如下图所示，也实现了jdbc的connector，可以通过jdbc 去操作数据库，但是flink-jdbc包中对数据库的操作是以ROW来操作并且对数据库事务的控制比较死板，有时候操作关系型数据库我们会非常怀念在java web应用开发中的非常优秀的mybatis框架，那么其实flink中是可以自己集成mybatis进来的。 我们这里以flink 1.9版本为例来进行集成。
如下图为flink内部自带的flink-jdbc：
创建一个flink的流式处理项目，引入flink的maven依赖和mybatis依赖（注意这里引入的是非spring版本，也就是mybatis的单机版）：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
<properties>
<flink.version>1.9.0</flink.version>
</properties>
<!-- https://mvnrepository.com/artifact/org.mybatis/mybatis -->
<dependency>
<groupId>org.mybatis</groupId>
<artifactId>mybatis</artifactId>
<version>3.5.2</version>
</dependency>
<!-- flink java 包 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>${flink.version}</version>
</dependency>
|
maven依赖引入以后，那么需要在resources下面定义mybatis-config.xml 配置：
mybatis-config.xml 需要定义如下配置：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
<typeAliases>
<typeAlias alias="BankBillPublic" type="xxxx.xx.xx.BankBillPublic" />
</typeAliases>
<environments default="development">
<environment id="development">
<transactionManager type="JDBC" />
<dataSource type="POOLED">
<property name="driver" value="com.mysql.jdbc.Driver" />
<property name="url" value="jdbc:mysql://xx.xx.xx.xx:3306/hue?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&autoReconnect=true" />
<property name="username" value="xxxx" />
<property name="password" value="xxxx*123%" />
</dataSource>
</environment>
</environments>
<mappers>
<mapper resource="mapper/xxxxxMapper.xml" />
</mappers>
</configuration>
|
typeAlias 标签中为自定义的数据类型，然后在xxxxxMapper.xml 中parameterType或者resultType就可以直接用这种定义的数据类型。
dataSource type="POOLED" 我们使用的是mybatis中的POOLED 类型，也就是连接池的方式去使用。默认支持如下这三种类型。
我们也可以使用阿里巴巴开源的druid连接池，那么就需要引入对应的maven依赖，如下所示：
|
1
2
3
4
5
|
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.0.14</version>
</dependency>
|
然后定义一个对应的druid的DataSource，如下所示：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
import java.sql.SQLException;
import java.util.Properties;
import javax.sql.DataSource;
import org.apache.ibatis.datasource.DataSourceFactory;
import com.alibaba.druid.pool.DruidDataSource;
public class DruidDataSourceFactory implements DataSourceFactory {
private Properties props;
@Override
public DataSource getDataSource() {
DruidDataSource dds = new DruidDataSource();
dds.setDriverClassName(this.props.getProperty("driver"));
dds.setUrl(this.props.getProperty("url"));
dds.setUsername(this.props.getProperty("username"));
dds.setPassword(this.props.getProperty("password"));
// 其他配置可以根据MyBatis主配置文件进行配置
try {
dds.init();
} catch (SQLException e) {
e.printStackTrace();
}
return dds;
}
@Override
public void setProperties(Properties props) {
this.props = props;
}
}
|
之后就可以mybatis的配置中使用了，如下所示：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
<typeAliases>
<typeAlias alias="BankBillPublic" type="xxxx.xx.xx.BankBillPublic" />
<typeAlias alias="DRUID"
type="com.xx.mybatis.druid.utils.DruidDataSourceFactory" />
</typeAliases>
<environments default="development">
<environment id="development">
<transactionManager type="JDBC" />
<dataSource type="DRUID">
<property name="driver" value="com.mysql.jdbc.Driver" />
<property name="url" value="jdbc:mysql://xx.xx.xx.xx:3306/hue?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&autoReconnect=true" />
<property name="username" value="xxxx" />
<property name="password" value="xxxx*123%" />
</dataSource>
</environment>
</environments>
<mappers>
<mapper resource="mapper/xxxxxMapper.xml" />
</mappers>
</configuration>
|
<mappers> 下面为定义的mybatis 的xxxxxMapper文件。里面放置的都是sql语句。
本文作者张永清，转载请注明出处：flink 流式处理中如何集成mybatis框架
xxxxxMapper.xml 中的sql示例：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="xx.xx.bigdata.flink.xx.xx.mapper.UserRelaInfoMapper">
<!--查询关键字匹配 -->
<select id="queryUserRelaInfo" parameterType="String" resultType="UserRelaInfo">
SELECT id AS id,
USER_NAME AS userName,
APPL_IDCARD AS applIdCard,
PEER_USER AS peerUser,
RELA_TYPE AS relaType,
CREATE_USER AS createUser,
CREATE_TIME AS createTime
FROM USER_RELA_INFO
<where>
<if test="applIdCard != null">
APPL_IDCARD=#{applIdCard}
</if>
<if test="peerUser != null">
AND PEER_USER=#{peerUser}
</if>
</where>
</select>
</mapper>
|
定义Mapper，一般可以定义一个interface ，和xxxxxMapper.xml中的namespace保持一致
注意传入的参数一般加上@Param 注解，传入的参数和xxxxxMapper.xml中需要的参数保持一致
|
1
2
3
|
public interface UserRelaInfoMapper {
List<UserRelaInfo> queryUserRelaInfo(@Param("applIdCard")String applIdCard,@Param("peerUser") String peerUser);
}
|
定义SessionFactory工厂（单例模式）：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
/**
*
* sqlsession factory 单例 事务设置为手动提交
*/
public class MybatisSessionFactory {
private static final Logger LOG = LoggerFactory.getLogger(MybatisSessionFactory.class);
private static SqlSessionFactory sqlSessionFactory;
private MybatisSessionFactory(){
super();
}
public synchronized static SqlSessionFactory getSqlSessionFactory(){
if(null==sqlSessionFactory){
InputStream inputStream=null;
try{
inputStream = MybatisSessionFactory.class.getClassLoader().getResourceAsStream("mybatis-config.xml");
sqlSessionFactory = new SqlSessionFactoryBuilder().build(inputStream);
}
catch (Exception e){
LOG.error("create MybatisSessionFactory read mybatis-config.xml cause Exception",e);
}
if(null!=sqlSessionFactory){
LOG.info("get Mybatis sqlsession sucessed....");
}
else {
LOG.info("get Mybatis sqlsession failed....");
}
}
return sqlSessionFactory;
}
}
|
使用mybatis 对数据库进行操作：
|
1
2
3
4
5
6
7
8
9
10
11
|
SqlSession sqlSession = MybatisSessionFactory.getSqlSessionFactory().openSession();
UserRelaInfoMapper userRelaInfoMapper = sqlSession.getMapper(UserRelaInfoMapper .class);
//调用对应的方法
userRelaInfoMapper.xxxx();
//提交事务
sqlSession.commit();
//回滚事务，一般可以捕获异常，在发生Exception的时候，事务进行回滚
sqlSession.rollback();
|
这里以mysql为示例，写一个flink下mysql的sink示例，可以自己来灵活控制事务的提交：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
public class MysqlSinkFunction<IN> extends RichSinkFunction {
private static final Logger LOG = LoggerFactory.getLogger(MysqlSinkFunction.class);
@Override
public void invoke(Object value, Context context) throws Exception{
SqlSession sqlSession = MybatisSessionFactory.getSqlSessionFactory().openSession();
try{
//插入
LOG.info("MysqlSinkFunction start to do insert data...");
xxx.xxx()；
//更新
LOG.info("MysqlSinkFunction start to do update data...");
xxx.xxx()；
//删除
LOG.info("MysqlSinkFunction start to do delete data...");
xxx.xxx()；
sqlSession.commit();
LOG.info("MysqlSinkFunction commit transaction success...");
}
catch (Throwable e){
sqlSession.rollback();
LOG.error("MysqlSinkFunction cause Exception,sqlSession transaction rollback...",e);
}
}
}
|
相信您如果以前在spring中用过mybatis的话，对上面的这些操作一定不会陌生。由此你也可以发现，在大数据中可以完美的集成mybatis，这样可以发挥mybatis框架对数据库操作的优势，使用起来也非常简单方便。
一旦集成了mybaitis后，在flink中就可以方便的对各种各样的关系型数据库进行操作了。
本文作者张永清，转载请注明出处：flink 流式处理中如何集成mybatis框架
flink 流式处理中如何集成mybatis框架相关推荐
- Flink流式计算从入门到实战 四
文章目录 六.Flink Table API 和Flink SQL 1.Table API和SQL是什么? 2.如何使用Table API 3.基础编程框架 3.1 创建TableEnvironmen ...
- Flink 流式计算在节省资源方面的简单分析
本文由小米的王加胜同学分享,文章介绍了 Apache Flink 在小米的发展,从 Spark Streaming 迁移到 Flink ,在调度计算与调度数据.Mini batch 与 streami ...
- Flink流式处理百万数据量CSV文件
前言 最近公司让做一个'没有必要'的需求 需求针对的对象 这是同一个csv文件的不同展示形式 Excel展示形式 文本展示形式 这个csv文件中可能有数百万条数据 需求 将所有的异常数据检测出来 什么 ...
- Flink流式计算从入门到实战 三
文章目录 四.Flink DataStream API 1.Flink程序的基础运行模型 2.Environment 运行环境 3.Source 3.1 基于File的数据源 3.2 基于Socket ...
- Flink流式计算从入门到实战 二
文章目录 三.Flink运行架构 1.JobManager和TaskManager 2.并发度与Slots 3.开发环境搭建 4.提交到集群执行 5.并行度分析 6.Flink整体运行流程 Flink ...
- Flink系列之Flink 流式编程模式总结
title: Flink系列 一.Flink 流式编程模式总结 1.1 基础总结 官网: https://flink.apache.org/ Apache Flink® - Stateful Comp ...
- Flink系列之Flink流式计算引擎基础理论
声明: 文章中代码及相关语句为自己根据相应理解编写,文章中出现的相关图片为自己实践中的截图和相关技术对应的图片,若有相关异议,请联系删除.感谢.转载请注明出处,感谢. By luoye ...
- 大数据之Flink流式计算引擎
Flink DataFlow 数据的分类 有界数据 离线的计算 效率:硬件相同的情况下:时间 无界数据 实时的计算 效率:硬件环境相同的情况下,吞吐量:数据处理的数量.延迟:结果输出的时间-数据接收的 ...
- Flink流式计算从入门到实战 一
文章目录 一.理解Flink与流计算 1.初识Flink 2.Flink的适用场景 3.流式计算梳理 二.Flink安装部署 1.Flink的部署方式 2.获取Flink 3.实验环境与前置软件 4. ...
最新文章
- 那些值得回味的MySQL的基础知识
- Linux软件安装常用方法
- [NOI2001]食物链
- 软件开发不是众人拾柴——《人月神话》读书笔记
- C语言程序设计——设计一个学生管理系统（完美运行的程序(●‘◡‘●)）
- yara 模式匹配 android,YARA——恶意软件模式匹配利器
- Binary Search二分法搜索C++程序
- linux opencv gtk 没窗口,OpenCV GTK+2.x error
- 终于解决了无运行无网上邻居的问题
- requests库学习
- vision画流程图的软件_这个可以代替Visio的流程图绘制软件，你值得拥有，还有网页版的~...
- hp 服务器 无线网卡,换个网卡再战！—记惠普4431s笔记本更换无线网卡体验
- wpf 复制到粘贴板_将WPF UI单元复制到剪贴板-阿里云开发者社区
- 网站打开速度慢的解决方法
- 表格识别-pytorch
- MPI聚合通信之MPI_Bcast函数
- 向量数据库入坑：使用 Docker 和 Milvus 快速构建本地轻量图片搜索引擎
- C++基本语法知识查漏补缺（一）
- 个人赛 A 题 传球游戏（ball）
- python学习：键盘输入一个或多个城市名，转换为拼音列表
热门文章
- python实现FM算法
- mysql主从同步简单原理_Mysql主从同步原理
- uniapp如何获取IP地址
- python自动填写excel,使用python自动填充语料关键词到Excel
- html怎么校验用户名重复,layui的表单验证支持ajax判断用户名是否重复的实例
- 常用语言注释及其快捷键
- 利用PHP从大txt文件中提取章节名称及行号
- 【机器学习｜数学基础】Mathematics for Machine Learning系列之图论（10）：匹配基本定理
- C++ 编程的基本形式：文件名、防卫式声明、头文件布局
- 远程监控摄像头在流媒体服务器显示无视频信号？
