上一篇文章我們通過自定義Source和Sink來實作了Flume的資料采集，本文我們再來看一個自定義MySQLSource的實體，關注專欄《破繭成蝶——大資料篇》，查看更多相關的內容~
目錄
一、需求說明
二、編碼實作
三、撰寫Flume組態檔
四、新建mysql表
五、測驗自定義的MySQLSource
一、需求說明
實時監控MySQL，從MySQL中獲取資料實時的傳輸到控制臺，
二、編碼實作
2.1 首先需要匯入相關的依賴包，如下所示，
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-core</artifactId>
<version>1.7.0</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.27</version>
</dependency>2.2 在classpath下新建jdbc.properties和log4j.properties，
（1）jdbc.properties
dbDriver=com.mysql.jdbc.Driver
dbUrl=jdbc:mysql://master:3306/xzw?useUnicode=true&characterEncoding=utf-8
dbUser=username
dbPassword=password
（2）log4j.properties
log4j.rootLogger=info,myconsole,myfile
log4j.appender.myconsole=org.apache.log4j.ConsoleAppender
log4j.appender.myconsole.layout=org.apache.log4j.SimpleLayout
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
2.3 撰寫SQL Source Parse，
package com.xzw.utils;
import org.apache.flume.Context;
import org.apache.flume.conf.ConfigurationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
/**
* @author: xzw
* @create_date: 2021/1/20 14:32
* @desc: SQL Source 解釋器
* @modifier:
* @modified_date:
* @desc:
*/
public class SQLSourceParse {
private static final Logger LOG = LoggerFactory.getLogger(SQLSourceParse.class);
private int runQueryDelay, //兩次查詢的時間間隔
startFrom, //開始id
currentIndex, //當前id
recordSixe = 0, //每次查詢回傳結果的條數
maxRow; //每次查詢的最大條數
private String table, //要操作的表
columnsToSelect, //要查詢的列
customQuery, //查詢陳述句
query, //構建查詢陳述句
defaultCharsetResultSet; //編碼集
//背景關系，用來獲取組態檔
private Context context;
//為定義的變數賦值，可在Flume任務的組態檔中修改
private static final int DEFAULT_QUERY_DELAY = 10000;
private static final int DEFAULT_START_VALUE = 0;
private static final int DEFAULT_MAX_ROWS = 2000;
private static final String DEFAULT_COLUMNS_SELECT = "*";
private static final String DEFAULT_CHARSET_RESULTSET = "UTF-8";
private static Connection conn = null;
private static PreparedStatement ps = null;
private static String connectionURL, connectionUserName, connectionPassword;
/**
* 加載靜態資源
*/
static {
Properties properties = new Properties();
try {
properties.load(SQLSourceParse.class.getClassLoader().getResourceAsStream("jdbc.properties"));
connectionURL = properties.getProperty("dbUrl");
connectionUserName = properties.getProperty("dbUser");
connectionPassword = properties.getProperty("dbPassword");
Class.forName(properties.getProperty("dbDriver"));
} catch (IOException | ClassNotFoundException e) {
LOG.error(e.toString());
}
}
/**
* 獲取jdbc連接
*
* @param url
* @param user
* @param password
* @return
*/
private static Connection InitConnection(String url, String user, String password) {
try {
Connection conn = DriverManager.getConnection(url, user, password);
if (conn == null)
throw new SQLException();
return conn;
} catch (SQLException e) {
e.printStackTrace();
}
return null;
}
/**
* 校驗相應的配置資訊
*
* @throws ConfigurationException
*/
private void checkProps() throws ConfigurationException {
if (table == null)
throw new ConfigurationException("table is not exist!");
if (connectionURL == null)
throw new ConfigurationException("connection url is null!");
if (connectionUserName == null)
throw new ConfigurationException("connection username is null!");
if (connectionPassword == null)
throw new ConfigurationException("connection password is null!");
}
/**
* 查詢一條資料
*
* @param sql
* @return
*/
private String queryOne(String sql) {
ResultSet resultSet = null;
try {
ps = conn.prepareStatement(sql);
resultSet = ps.executeQuery();
while (resultSet.next()) {
return resultSet.getString(1);
}
} catch (SQLException e) {
e.printStackTrace();
}
return null;
}
/**
* 獲取當前id的offset
*
* @param startFrom
* @return
*/
private Integer getStatusDBIndex(int startFrom) {
//從Flume的meta中獲取當前id是多少
String dbIndex = queryOne("select currentIndex from flume_meta where source_tab='" + table + "'");
if (dbIndex != null)
return Integer.parseInt(dbIndex);
//如果沒有資料，說明是第一次查詢或者資料表中還沒有存入資料，回傳最初傳入的值
return startFrom;
}
/**
* 構建SQL陳述句
*
* @return
*/
private String buildQuery() {
String sql = "";
//獲取當前的id
currentIndex = getStatusDBIndex(startFrom);
LOG.info(currentIndex + "");
if (customQuery == null) {
sql = "select " + columnsToSelect + " from " + table;
} else {
sql = customQuery;
}
StringBuilder execsql = new StringBuilder(sql);
//以id作為offset
if (!sql.contains("where")) {
execsql.append(" where ");
execsql.append(" id ").append(" > ").append(currentIndex);
return execsql.toString();
} else {
int length = execsql.toString().length();
return execsql.toString().substring(0, length - String.valueOf(currentIndex).length()) + currentIndex;
}
}
/**
* 構造方法
*
* @param context
*/
public SQLSourceParse(Context context) throws ConfigurationException {
//初始化背景關系
this.context = context;
//默認值引數，獲取Flume組態檔中的引數，讀取不到的使用默認值
this.columnsToSelect = context.getString("columns.to.select", DEFAULT_COLUMNS_SELECT);
this.runQueryDelay = context.getInteger("run.query.delay", DEFAULT_QUERY_DELAY);
this.startFrom = context.getInteger("start.from", DEFAULT_START_VALUE);
this.defaultCharsetResultSet = context.getString("default.charset.resultset", DEFAULT_CHARSET_RESULTSET);
//無默認值引數，獲取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);
//校驗相應的配置資訊，如果沒有默認值的引數也沒有賦值，拋出例外
checkProps();
//獲取當前的id
currentIndex = getStatusDBIndex(startFrom);
//構建查詢陳述句
query = buildQuery();
}
/**
* 執行查詢
*
* @return
*/
public List<List<Object>> execQuery() {
try {
//每次執行查詢都要重新生成SQL，因為id不同
customQuery = buildQuery();
//存放結果的集合
List<List<Object>> results = new ArrayList<>();
if (ps == null)
ps = conn.prepareStatement(customQuery);
ResultSet rs = ps.executeQuery(customQuery);
while (rs.next()) {
//存放一條資料的集合
List<Object> row = new ArrayList<>();
//將回傳結果放入集合
for (int i = 1; i < rs.getMetaData().getColumnCount(); i++) {
row.add(rs.getObject(i));
}
results.add(row);
}
LOG.info("execSql:" + customQuery + "\nresultSize:" + results.size());
return results;
} catch (SQLException e) {
LOG.error(e.toString());
// 重新連接
conn = InitConnection(connectionURL, connectionUserName, connectionPassword);
}
return null;
}
/**
* 將結果集轉化為字串，每一條資料是一個list集合，將每一個小的list集合轉化為字串
*
* @param queryResult
* @return
*/
public List<String> getAllRows(List<List<Object>> queryResult) {
List<String> allRows = new ArrayList<>();
if (queryResult == null || queryResult.isEmpty())
return allRows;
StringBuilder row = new StringBuilder();
for (List<Object> rawRow : queryResult) {
Object value = null;
for (Object aRawRow : rawRow) {
value = aRawRow;
if (value == null) {
row.append(",");
} else {
row.append(aRawRow.toString()).append(",");
}
}
allRows.add(row.toString());
row = new StringBuilder();
}
return allRows;
}
/**
* 執行sql陳述句
*
* @param sql
*/
private void execSql(String sql) {
try {
ps = conn.prepareStatement(sql);
LOG.info("exec::" + sql);
ps.execute();
} catch (SQLException e) {
e.printStackTrace();
}
}
/**
* 更新offset元資料狀態，每次回傳結果集后呼叫，必須記錄每次查詢的offset值，為程式中斷續跑資料時使用，以id為offset
*
* @param size
*/
public void updateOffset2DB(int size) {
//以source_tab做為KEY，如果不存在則插入，存在則更新（每個源表對應一條記錄）
String sql = "insert into flume_meta(source_tab,currentIndex) VALUES('"
+ this.table
+ "','" + (recordSixe += size)
+ "') on DUPLICATE key update source_tab=values(source_tab),currentIndex=values(currentIndex)";
LOG.info("updateStatus Sql:" + sql);
execSql(sql);
}
/**
* 關閉相關資源
*/
public void close() {
try {
ps.close();
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
public int getCurrentIndex() {
return currentIndex;
}
public void setCurrentIndex(int newValue) {
currentIndex = newValue;
}
public int getRunQueryDelay() {
return runQueryDelay;
}
public String getQuery() {
return query;
}
public String getConnectionURL() {
return connectionURL;
}
public boolean isCustomQuerySet() {
return (customQuery != null);
}
public Context getContext() {
return context;
}
public String getConnectionUserName() {
return connectionUserName;
}
public String getConnectionPassword() {
return connectionPassword;
}
public String getDefaultCharsetResultSet() {
return defaultCharsetResultSet;
}
}
2.4 撰寫MySQL Source，
package com.xzw.source;
import com.xzw.utils.SQLSourceParse;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.conf.ConfigurationException;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
/**
* @author: xzw
* @create_date: 2021/1/20 14:30
* @desc: 自定義的MySQL Source
* @modifier:
* @modified_date:
* @desc:
*/
public class MySQLSource extends AbstractSource implements Configurable, PollableSource {
//列印日志
private static final Logger LOG = LoggerFactory.getLogger(MySQLSource.class);
//定義parse
private SQLSourceParse sqlSourceParse;
@Override
public Status process() throws EventDeliveryException {
try {
//查詢資料表
List<List<Object>> result = sqlSourceParse.execQuery();
//存放event的集合
List<Event> events = new ArrayList<>();
//存放event頭
HashMap<String, String> header = new HashMap<>();
//如果沒有回傳資料，則將資料封裝為event
if (!result.isEmpty()) {
List<String> allRows = sqlSourceParse.getAllRows(result);
Event event = null;
for (String row :
allRows) {
event = new SimpleEvent();
event.setBody(row.getBytes());
event.setHeaders(header);
events.add(event);
}
//將event寫入channel
this.getChannelProcessor().processEventBatch(events);
//更新資料表中的offset資訊
sqlSourceParse.updateOffset2DB(result.size());
}
//等待時長
Thread.sleep(sqlSourceParse.getRunQueryDelay());
return Status.READY;
} catch (InterruptedException e) {
LOG.error("Error procesing row", e);
return Status.BACKOFF;
}
}
@Override
public long getBackOffSleepIncrement() {
return 0;
}
@Override
public long getMaxBackOffSleepInterval() {
return 0;
}
@Override
public void configure(Context context) {
try {
sqlSourceParse = new SQLSourceParse(context);
} catch (ConfigurationException e) {
e.printStackTrace();
}
}
@Override
public synchronized void stop() {
LOG.info("Stopping sql source {} ...", getName());
try {
//關閉資源
sqlSourceParse.close();
} finally {
super.stop();
}
}
public static void main(String[] args) {
}
}
三、撰寫Flume組態檔
撰寫flume-mysqlsource.conf組態檔，具體內容如下所示：
a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = com.xzw.source.MySQLSource
a1.sources.r1.connection.url = jdbc:mysql://192.168.0.82:3306/xzw
a1.sources.r1.connection.user = username
a1.sources.r1.connection.password = password
a1.sources.r1.table = people
a1.sources.r1.columns.to.select = *
#a1.sources.r1.custom.query = select * from people
#a1.sources.r1.incremental.column.name = id
#a1.sources.r1.incremental.value = 0
a1.sources.r1.run.query.delay=5000
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1四、新建mysql表
在MySQL中新建兩個表，分別如下所示：
CREATE TABLE `people` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(255) NOT NULL,
PRIMARY KEY (`id`)
);
CREATE TABLE `flume_meta` (
`source_tab` varchar(255) NOT NULL,
`currentIndex` varchar(255) NOT NULL,
PRIMARY KEY (`source_tab`)
);并向people表中插入資料：
insert into people values(1, 'xzw');
insert into people values(2, 'lzq');
insert into people values(3, 'yxy');
insert into people values(4, 'lyq');五、測驗自定義的MySQLSource
啟動Flume，進行測驗，命令如下：
bin/flume-ng agent --conf conf/ --name a1 --conf-file conf/flume-mysqlsource.conf -Dflume.root.logger=INFO,console
本文相當于是在上一篇文章的基礎上進行的一個簡單延伸，本文就到這里了，實作程序相對簡單，你們在此程序中遇到了什么問題，歡迎留言，讓我看看你們都遇到了哪些問題~
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/261871.html
標籤：其他