欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

flink 流式处理中如何集成mybatis框架

程序员文章站 2022-03-02 13:56:37
flink 中自身虽然实现了大量的connectors,如下图所示,也实现了jdbc的connector,可以通过jdbc 去操作数据库,但是flink-jdbc包中对数据库的操作是以ROW来操作并且对数据库事务的控制比较死板,有时候操作关系型数据库我们会非常怀念在java web应用开发中的非常优 ......

flink 中自身虽然实现了大量的connectors,如下图所示,也实现了jdbc的connector,可以通过jdbc 去操作数据库,但是flink-jdbc包中对数据库的操作是以row来操作并且对数据库事务的控制比较死板,有时候操作关系型数据库我们会非常怀念在java web应用开发中的非常优秀的mybatis框架,那么其实flink中是可以自己集成mybatis进来的。 我们这里以flink 1.9版本为例来进行集成。

flink 流式处理中如何集成mybatis框架

如下图为flink内部自带的flink-jdbc:

 flink 流式处理中如何集成mybatis框架

创建一个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 配置:

flink 流式处理中如何集成mybatis框架

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框架