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="https://www.cnblogs.com/laoqing/p/com.mysql.jdbc.Driver" /> <property name="url" value="https://www.cnblogs.com/laoqing/p/jdbc:mysql://xx.xx.xx.xx:3306/hue?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&autoReconnect=true" /> <property name="username" value="https://www.cnblogs.com/laoqing/p/xxxx" /> <property name="password" value="https://www.cnblogs.com/laoqing/p/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依賴，如下所示：
<dependency> <groupId>com.alibaba</groupId> <artifactId>druid</artifactId> <version>1.0.14</version> </dependency>
然后定義一個對應的druid的DataSource，如下所示：
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的配置中使用了，如下所示：
<?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="https://www.cnblogs.com/laoqing/p/com.mysql.jdbc.Driver" /> <property name="url" value="https://www.cnblogs.com/laoqing/p/jdbc:mysql://xx.xx.xx.xx:3306/hue?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&autoReconnect=true" /> <property name="username" value="https://www.cnblogs.com/laoqing/p/xxxx" /> <property name="password" value="https://www.cnblogs.com/laoqing/p/xxxx*123%" /> </dataSource> </environment> </environments> <mappers> <mapper resource="mapper/xxxxxMapper.xml" /> </mappers></configuration>
<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框架
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/44431.html
標籤：大數據
上一篇：頻繁項集的產生及經典演算法