实时监控MySQL,从MySQL中获取数据传输到HDFS或者其他存储框架,所以此时需要我们自己实现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获取配置信息。
1、导入pom依赖
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
MySQLSource.java
packagecom.atguigu.source;importorg.apache.flume.Context;importorg.apache.flume.Event;importorg.apache.flume.EventDeliveryException;importorg.apache.flume.PollableSource;importorg.apache.flume.conf.Configurable;importorg.apache.flume.event.SimpleEvent;importorg.apache.flume.source.AbstractSource;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importjava.text.ParseException;importjava.util.ArrayList;importjava.util.HashMap;importjava.util.List;public class MySQLSource extends AbstractSource implementsConfigurable, PollableSource {//打印日志 private static final Logger LOG = LoggerFactory.getLogger(MySQLSource.class);//定义sqlHelper privateSQLSourceHelper sqlSourceHelper; @Overridepublic longgetBackOffSleepIncrement() {return 0; } @Overridepublic longgetMaxBackOffSleepInterval() {return 0; } @Overridepublic voidconfigure(Context context) {try{//初始化 sqlSourceHelper = newSQLSourceHelper(context); }catch(ParseException e) { e.printStackTrace(); } } @Overridepublic Status process() throwsEventDeliveryException {try{//查询数据表 List> result =sqlSourceHelper.executeQuery();//存放event的集合 List
1、jar包准备
1)将mysql驱动包放入flume的lib目录下[kris@hadoop101 flume]$ cp /opt/software/mysql-connector-java-5.1.27/mysql-connector-java-5.1.27-bin.jar ./lib
2)打包项目并将jar包放入flume的lib目录下
2、配置文件准备
1)创建配置文件并打开
vim mysql.conf /# Name the components on this agent a1.sources=r1 a1.sinks=k1 a1.channels=c1 /# Describe/configure the source a1.sources.r1.type=com.atguigu.source.MySQLSource a1.sources.r1.connection.url= jdbc:mysql://192.168.1.101:3306/mysqlsource a1.sources.r1.connection.user =root a1.sources.r1.connection.password= 123456a1.sources.r1.table=student a1.sources.r1.columns.to.select = //#a1.sources.r1.incremental.column.name= id/#a1.sources.r1.incremental.value= 0a1.sources.r1.run.query.delay=5000/# Describe the sink a1.sinks.k1.type=logger* /# Describe the channel a1.channels.c1.type=memory a1.channels.c1.capacity= 1000a1.channels.c1.transactionCapacity= 100/# Bind the source and sink to the channel a1.sources.r1.channels=c1 a1.sinks.k1.channel= c1
3、mysql表准备
1)创建mysqlsource数据库
CREATE DATABASE mysqlsource;
2)在mysqlsource数据库下创建数据表student和元数据表flume_metaCREATE DATABASEmysqlsource;USEmysqlsource;CREATE TABLEstudent( idINT(11) NOT NULLAUTOINCREMENT, name
VARCHAR(255) NOT NULL,PRIMARY KEY(id));CREATE TABLEflumemeta ( sourcetabVARCHAR(255) NOT NULL, currentIndexVARCHAR(255) NOT NULL,PRIMARY KEY(sourcetab));
3) 向数据表中添加数据
1kris2egon3 alex
5、测试并查看结果
1) 任务执行[kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -na1-f job/mysql.conf -Dflume.root.logger=INFO,console
2) 结果展示,如图所示:
往studnet表中更新数据: