我要评论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的单机版）：
<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 需要定义如下配置：
<?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就可以直接用这种定义的数据类型。
<mappers> 下面为定义的mybatis 的xxxxxmapper文件。里面放置的都是sql语句。
本文作者张永清，转载请注明出处：flink 流式处理中如何集成mybatis框架
xxxxxmapper.xml 中的sql示例：
<?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中需要的参数保持一致
public interface userrelainfomapper {
list<userrelainfo> queryuserrelainfo(@param("applidcard")string applidcard,@param("peeruser") string peeruser);
}
定义sessionfactory工厂（单例模式）：
/**
*
* 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 对数据库进行操作：
sqlsession sqlsession = mybatissessionfactory.getsqlsessionfactory().opensession();
userrelainfomapper userrelainfomapper = sqlsession.getmapper(userrelainfomapper .class);
//调用对应的方法
userrelainfomapper.xxxx();
//提交事务
sqlsession.commit();
//回滚事务，一般可以捕获异常，在发生exception的时候，事务进行回滚
sqlsession.rollback();
这里以mysql为示例，写一个flink下mysql的sink示例，可以自己来灵活控制事务的提交：
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框架
您可能感兴趣的文章:
如您对本文有疑问或者有任何想说的，请点击进行留言回复，万千网友为您解惑！
网友评论