MySQLSource-Flume

1. 自定义Source说明

实时监控MySQL，从MySQL中获取数据传输到HDFS或者其他存储框架，所以此时需要我们自己实现MySQLSource。

2. 自定义MySQLSource步骤

根据官方说明自定义mysqlsource需要继承AbstractSource类并实现Configurable和PollableSource接口。

实现相应方法：

getBackOffSleepIncrement()//暂不用

getMaxBackOffSleepInterval()//暂不用

configure(Context context)//初始化context

process()//获取数据（从mysql获取数据，业务处理比较复杂，所以我们定义一个专门的类——SQLSourceHelper来处理跟mysql的交互），封装成event并写入channel，这个方法被循环调用

stop()//关闭相关的资源

PollableSource：从source中提取数据，将其发送到channel。

Configurable：实现了Configurable的任何类都含有一个context，使用context获取配置信息。

3. 代码实现

1、导入pom依赖 org.apache.maven.plugins maven-compiler-plugin 8 /#JDK版本 8 org.apache.flume flume-ng-core 1.7.0 mysql mysql-connector-java 5.1.27

2、添加配置信息

在classpath下添加jdbc.properties和log4j.properties

jdbc.propertiesdbDriver=com.mysql.jdbc.Driver dbUrl=jdbc:mysql://hadoop101:3306/mysqlsource?useUnicode=true&characterEncoding=utf-8 dbUser=root dbPassword=123456

log4j.properties

/#--------console----------- log4j.rootLogger=info,myconsole,myfile log4j.appender.myconsole=org.apache.log4j.ConsoleAppender log4j.appender.myconsole.layout=org.apache.log4j.SimpleLayout /#log4j.appender.myconsole.layout.ConversionPattern =%d [%t] %-5p [%c] - %m%n /#log4j.rootLogger=error,myfile log4j.appender.myfile=org.apache.log4j.DailyRollingFileAppender log4j.appender.myfile.File=/tmp/flume.log log4j.appender.myfile.layout=org.apache.log4j.PatternLayout log4j.appender.myfile.layout.ConversionPattern =%d [%t] %-5p [%c] - %m%n

3、SQLSourceHelper

1)属性说明：属性 说明（括号中为默认值） runQueryDelay 查询时间间隔（10000） batchSize 缓存大小（100） startFrom 查询语句开始id（0） currentIndex 查询语句当前id，每次查询之前需要查元数据表 recordSixe 查询返回条数 table 监控的表名 columnsToSelect 查询字段（/*） customQuery 用户传入的查询语句 query 查询语句 defaultCharsetResultSet 编码格式（UTF-8）

2）方法说明：

2) 方法说明： 方法 说明 SQLSourceHelper(Context context) 构造方法，初始化属性及获取JDBC连接 InitConnection(String url, String user, String pw) 获取JDBC连接 checkMandatoryProperties() 校验相关属性是否设置（实际开发中可增加内容） buildQuery() 根据实际情况构建sql语句，返回值String executeQuery() 执行sql语句的查询操作，返回值List> getAllRows(List> queryResult) 将查询结果转换为String，方便后续操作 updateOffset2DB(int size) 根据每次查询结果将offset写入元数据表 execSql(String sql) 具体执行sql语句方法 getStatusDBIndex(int startFrom) 获取元数据表中的offset queryOne(String sql) 获取元数据表中的offset实际sql语句执行方法 close() 关闭资源

3)  代码实现：

SQLSourceHelper .java

packagecom.atguigu.source;importorg.apache.flume.Context;importorg.apache.flume.conf.ConfigurationException;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importjava.io.IOException;import java.sql./;importjava.text.ParseException;importjava.util.ArrayList;importjava.util.List;importjava.util.Properties;public classSQLSourceHelper {private static final Logger LOG = LoggerFactory.getLogger(SQLSourceHelper.class);private int runQueryDelay, //两次查询的时间间隔 startFrom, //开始id currentIndex, //当前id recordSixe = 0, //每次查询返回结果的条数 maxRow; //每次查询的最大条数 private String table, //要操作的表 columnsToSelect, //用户传入的查询的列 customQuery, //用户传入的查询语句 query, //构建的查询语句 defaultCharsetResultSet;//编码集//上下文，用来获取配置文件 privateContext context;//为定义的变量赋值（默认值），可在flume任务的配置文件中修改 private static final int DEFAULTQUERYDELAY = 10000;private static final int DEFAULTSTARTVALUE = 0;private static final int DEFAULTMAXROWS = 2000;private static final String DEFAULTCOLUMNSSELECT = "/";private static final String DEFAULTCHARSETRESULTSET = "UTF-8";private static Connection conn = null;private static PreparedStatement ps = null;private staticString connectionURL, connectionUserName, connectionPassword;//加载静态资源 static{ Properties p= newProperties();try{ p.load(SQLSourceHelper.class.getClassLoader().getResourceAsStream("jdbc.properties")); connectionURL= p.getProperty("dbUrl"); connectionUserName= p.getProperty("dbUser"); connectionPassword= p.getProperty("dbPassword"); Class.forName(p.getProperty("dbDriver")); }catch (IOException |ClassNotFoundException e) { LOG.error(e.toString()); } }//获取JDBC连接 private staticConnection InitConnection(String url, String user, String pw) {try{ Connection conn=DriverManager.getConnection(url, user, pw);if (conn == null)throw newSQLException();returnconn; }catch(SQLException e) { e.printStackTrace(); }return null; }//构造方法 SQLSourceHelper(Context context) throwsParseException {//初始化上下文 this.context =context;//有默认值参数：获取flume任务配置文件中的参数，读不到的采用默认值 this.columnsToSelect = context.getString("columns.to.select", DEFAULTCOLUMNSSELECT);this.runQueryDelay = context.getInteger("run.query.delay", DEFAULTQUERYDELAY);this.startFrom = context.getInteger("start.from", DEFAULTSTARTVALUE);this.defaultCharsetResultSet = context.getString("default.charset.resultset", DEFAULTCHARSETRESULTSET);//无默认值参数：获取flume任务配置文件中的参数 this.table = context.getString("table");this.customQuery = context.getString("custom.query"); connectionURL= context.getString("connection.url"); connectionUserName= context.getString("connection.user"); connectionPassword= context.getString("connection.password"); conn=InitConnection(connectionURL, connectionUserName, connectionPassword);//校验相应的配置信息，如果没有默认值的参数也没赋值，抛出异常 checkMandatoryProperties();//获取当前的id currentIndex =getStatusDBIndex(startFrom);//构建查询语句 query =buildQuery(); }//校验相应的配置信息（表，查询语句以及数据库连接的参数） private voidcheckMandatoryProperties() {if (table == null) {throw new ConfigurationException("property table not set"); }if (connectionURL == null) {throw new ConfigurationException("connection.url property not set"); }if (connectionUserName == null) {throw new ConfigurationException("connection.user property not set"); }if (connectionPassword == null) {throw new ConfigurationException("connection.password property not set"); } }//构建sql语句 privateString buildQuery() { String sql= "";//获取当前id currentIndex =getStatusDBIndex(startFrom); LOG.info(currentIndex+ "");if (customQuery == null) { sql= "SELECT " + columnsToSelect + " FROM " +table; }else{ sql=customQuery; } StringBuilder execSql= newStringBuilder(sql);//以id作为offset if (!sql.contains("where")) { execSql.append(" where "); execSql.append("id").append(">").append(currentIndex);returnexecSql.toString(); }else{int length =execSql.toString().length();return execSql.toString().substring(0, length - String.valueOf(currentIndex).length()) +currentIndex; } }//执行查询 List>executeQuery() {try{//每次执行查询时都要重新生成sql，因为id不同 customQuery =buildQuery();//存放结果的集合 List> results = new ArrayList<>();if (ps == null) {// ps =conn.prepareStatement(customQuery); } ResultSet result=ps.executeQuery(customQuery);while(result.next()) {//存放一条数据的集合（多个列） List