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

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語句

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的話，對上面的這些操做必定不會陌生。
本文做者張永清，轉載請註明出處：flink 流式處理中如何集成mybatis框架