Flume案例Ganglia监控

小鱼儿 2022-04-25 01:48 236阅读 0赞

Flume案例和Flume监控系统的使用：

## 安装 ##

1. 将apache-flume-1.7.0-bin.tar.gz上传到linux的/opt/software目录下
2. 解压apache-flume-1.7.0-bin.tar.gz到/opt/module/目录下
 
 <table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 software]$ tar -zxf apache-flume-1.7.0-bin.tar.gz -C /opt/module/ </pre></td> 
 </tr> 
 </tbody> 
 </table>
3. 修改apache-flume-1.7.0-bin的名称为flume

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 module]$ mv apache-flume-1.7.0-bin flume </pre></td> 
 </tr> 
 </tbody> 
</table>

1.  将flume/conf下的flume-env.sh.template文件修改为flume-env.sh，并配置flume-env.sh文件

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 </pre></td> 
 <td><pre>[hadoop@datanode1 module]$ mv flume-env.sh.template flume-env.sh [hadoop@datanode1 module]$ vi flume-env.sh export JAVA_HOME=/opt/module/jdk1.8.0_162 </pre></td> 
 </tr> 
 </tbody> 
</table>

## 案例实操 ##

### 监控端口数据 ###

案例需求：首先，Flume监控本机44444端口，然后通过telnet工具向本机44444端口发送消息，最后Flume将监听的数据实时显示在控制台。

判断端口是否被占用

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>sudo netstat -tunlp | grep 44444 </pre></td> 
 </tr> 
 </tbody> 
</table>

功能描述：netstat命令是一个监控TCP/IP网络的非常有用的工具，它可以显示路由表、实际的网络连接以及每一个网络接口设备的状态信息。

基本语法：netstat \[选项\]

选项参数：

-t或—tcp：显示TCP传输协议的连线状况；

-u或—udp：显示UDP传输协议的连线状况；

-n或—numeric：直接使用ip地址，而不通过域名服务器；

-l或—listening：显示监控中的服务器的Socket；

-p或—programs：显示正在使用Socket的程序识别码和程序名称；

#### 配置 ####

#### 启动 ####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-telnet-logger.conf -Dflume.root.logger=INFO,console 参数说明： 	--conf conf/ ：表示配置文件存储在conf/目录 	--name a1	：表示给agent起名为a1 	--conf-file job/flume-telnet.conf ：flume本次启动读取的配置文件是在job文件夹下的flume-telnet.conf文件。 	-Dflume.root.logger==INFO,console ：-D表示flume运行时动态修改flume.root.logger参数属性值，并将控制台日志打印级别设置为INFO级别。日志级别包括:log、info、warn、error。 </pre></td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>telnet localhost 44444 </pre></td> 
 </tr> 
 </tbody> 
</table>

![Fb1qot.png][]

### 实时读取本地文件到HDFS案例 ###

#### 测试脚本 ####

#### flume-file-hdfs.conf ####

![Fb0ndU.png][]

### 实时读取目录文件到HDFS案例 ###

#### 需求分析 ####

![Fb05Ss.png][]

#### 配置 ####

#### 测试脚本 ####

#### 启动 ####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-dir-hdfs.conf </pre></td> 
 </tr> 
 </tbody> 
</table>

#### 注意 ####

1.  在使用Spooling Directory Source时
2.  不要在监控目录中创建并持续修改文件
3.  上传完成的文件会以.COMPLETED结尾
4.  被监控文件夹每600毫秒扫描一次文件变动

#### 查看 ####

![Fb2PFP.png][]

### 查看本地文件 ###

![Fb2uoq.png][]

### 单数据源多出口案例(一) ###

![FSNdhR.png][]

#### 分析 ####

案例需求：使用flume-1监控文件变动，flume-1将变动内容传递给flume-2，flume-2负责存储到HDFS。同时flume-1将变动内容传递给flume-3，flume-3负责输出到local filesystem。

需求分析：

![Fb2Gy4.png][]

#### 步骤 ####

1. 在/opt/module/flume/job目录下创建group1文件夹
 
 <table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 job]$ cd group1/ </pre></td> 
 </tr> 
 </tbody> 
 </table>
 
 1. 在datanode3节点上/opt/module/datas/目录下创建flume3文件夹
 
 <table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode3 datas]$ mkdir flume3/ </pre></td> 
 </tr> 
 </tbody> 
 </table>
2. 配置1个接收日志文件的source和两个channel、两个sink，分别输送给flume-flume-hdfs和flume-flume-dir。
 
 datanode1配置文件
 
 <table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 </pre></td> 
 <td><pre>[hadoop@datanode1 group1]$ vim flume-file-flume.conf # Name the components on this agent a1.sources = r1 a1.sinks = k1 k2 a1.channels = c1 c2 # 将数据流复制给多个channel a1.sources.r1.selector.type = replicating # Describe/configure the source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /opt/module/datas/logs.log a1.sources.r1.shell = /bin/bash -c # Describe the sink a1.sinks.k1.type = avro a1.sinks.k1.hostname = datanode2 a1.sinks.k1.port = 4141 a1.sinks.k2.type = avro a1.sinks.k2.hostname = datanode3 a1.sinks.k2.port = 4142 # Describe the channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 a1.channels.c2.type = memory a1.channels.c2.capacity = 1000 a1.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 c2 a1.sinks.k1.channel = c1 a1.sinks.k2.channel = c2 </pre></td> 
 </tr> 
 </tbody> 
 </table>
 
 datanode2配置文件
 
 <table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 </pre></td> 
 <td><pre>[hadoop@datanode2 group1]$ vim flume-flume-hdfs.conf # Name the components on this agent a2.sources = r1 a2.sinks = k1 a2.channels = c1 # Describe/configure the source a2.sources.r1.type = avro a2.sources.r1.bind = datanode2 a2.sources.r1.port = 4141 # Describe the sink a2.sinks.k1.type = hdfs a2.sinks.k1.hdfs.path = hdfs://datanode1:9000/flume2/%Y%m%d/%H #上传文件的前缀 a2.sinks.k1.hdfs.filePrefix = flume2- #是否按照时间滚动文件夹 a2.sinks.k1.hdfs.round = true #多少时间单位创建一个新的文件夹 a2.sinks.k1.hdfs.roundValue = 1 #重新定义时间单位 a2.sinks.k1.hdfs.roundUnit = hour #是否使用本地时间戳 a2.sinks.k1.hdfs.useLocalTimeStamp = true #积攒多少个Event才flush到HDFS一次 a2.sinks.k1.hdfs.batchSize = 100 #设置文件类型，可支持压缩 a2.sinks.k1.hdfs.fileType = DataStream #多久生成一个新的文件 a2.sinks.k1.hdfs.rollInterval = 600 #设置每个文件的滚动大小大概是128M a2.sinks.k1.hdfs.rollSize = 134217700 #文件的滚动与Event数量无关 a2.sinks.k1.hdfs.rollCount = 0 #最小冗余数 a2.sinks.k1.hdfs.minBlockReplicas = 1 # Describe the channel a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 100 </pre></td> 
 </tr> 
 </tbody> 
 </table>
 
 datanode3配置文件
 
 <table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 </pre></td> 
 <td><pre>[hadoop@datanode3 group1]$ vim flume-flume-dir.conf me the components on this agent a3.sources = r1 a3.sinks = k1 a3.channels = c2 # Describe/configure the source a3.sources.r1.type = avro a3.sources.r1.bind = datanode3 a3.sources.r1.port = 4142 # Describe the sink a3.sinks.k1.type = file_roll a3.sinks.k1.sink.directory = /opt/module/datas/flume3 # Describe the channel a3.channels.c2.type = memory a3.channels.c2.capacity = 1000 a3.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a3.sources.r1.channels = c2 a3.sinks.k1.channel = c2启动 </pre></td> 
 </tr> 
 </tbody> 
 </table>

#### 启动 ####

datanode1

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group1/flume-file-flume.conf </pre></td> 
 </tr> 
 </tbody> 
</table>

datanode2

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode2 flume]$ bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group1/flume-flume-hdfs.conf </pre></td> 
 </tr> 
 </tbody> 
</table>

datanode3

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode3 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group1/flume-flume-dir.conf </pre></td> 
 </tr> 
 </tbody> 
</table>

### 单数据源多出口案例(二) ###

![FSN9kd.png][]

#### 需求 ####

案例需求：使用flume-1监控文件变动，flume-1将变动内容传递给flume-2，flume-2负责存储到HDFS。同时flume-1将变动内容传递给flume-3，flume-3也负责存储到HDFS

![FbzfSO.png][]

#### 实现 ####

datanode1

datanode2

datanode3

#### 启动 ####

datanode1

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group1/flume-netcat-flume.conf </pre></td> 
 </tr> 
 </tbody> 
</table>

datanode2

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode2 flume]$ bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group1/flume-flume1.conf -Dflume.root.logger=INFO,console </pre></td> 
 </tr> 
 </tbody> 
</table>

datanod3

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode3 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group1/flume-flume2.conf -Dflume.root.logger=INFO,console </pre></td> 
 </tr> 
 </tbody> 
</table>

![Fq9gkn.png][]

### 多数据源汇总案例 ###

#### 需求 ####

![FSN110.png][]

datanode1上的flume-1监控一个软件的log日志，

datanode2上的flume-2监控某一个端口的数据流，

flume-1与flume-2将数据发送给datanode3上的flume-3，flume-3将最终数据打印到控制台

#### 步骤 ####

1.  分发flume

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 </pre></td> 
 <td><pre>[hadoop@datanode2 job]$ mkdir group2 [hadoop@datanode2 job]$ xsync /opt/module/flume/ </pre></td> 
 </tr> 
 </tbody> 
</table>

1.  datanode1配置source用于监控hive.log文件，配置sink输出数据到下一级flume。

#### 启动 ####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 </pre></td> 
 <td><pre>[hadoop@datanode3 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group2/flume3.conf -Dflume.root.logger=INFO,console [hadoop@datanode2 flume]$ bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group2/flume2.conf [hadoop@datanode1 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group2/flume1.conf </pre></td> 
 </tr> 
 </tbody> 
</table>

![FqFh59.png][]

### 自定义MYSQLSource ###

#### SQLSourceHelper ####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 </pre></td> 
 <td><pre>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.text.ParseException; import java.util.ArrayList; import java.util.List; import java.util.Properties; public class SQLSourceHelper {
 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;//编码集 //上下文，用来获取配置文件 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 p = new Properties(); 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 static Connection InitConnection(String url, String user, String pw) {
 try {
 Connection conn = DriverManager.getConnection(url, user, pw); if (conn == null) throw new SQLException(); return conn; } catch (SQLException e) {
 e.printStackTrace(); } return null; } //构造方法 SQLSourceHelper(Context context) throws ParseException {
 //初始化上下文 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); //校验相应的配置信息，如果没有默认值的参数也没赋值，抛出异常 checkMandatoryProperties(); //获取当前的id currentIndex = getStatusDBIndex(startFrom); //构建查询语句 query = buildQuery(); } //校验相应的配置信息（表，查询语句以及数据库连接的参数） private void checkMandatoryProperties() {
 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语句 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("&gt;").append(currentIndex); return execSql.toString(); } else {
 int length = execSql.toString().length(); return execSql.toString().substring(0, length - String.valueOf(currentIndex).length()) + currentIndex; } } //执行查询 List&lt;List&lt;Object&gt;&gt; executeQuery() {
 try {
 //每次执行查询时都要重新生成sql，因为id不同 customQuery = buildQuery(); //存放结果的集合 List&lt;List&lt;Object&gt;&gt; results = new ArrayList&lt;&gt;(); if (ps == null) {
 // ps = conn.prepareStatement(customQuery); } ResultSet result = ps.executeQuery(customQuery); while (result.next()) {
 //存放一条数据的集合（多个列） List&lt;Object&gt; row = new ArrayList&lt;&gt;(); //将返回结果放入集合 for (int i = 1; i &lt;= result.getMetaData().getColumnCount(); i++) {
 row.add(result.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集合转化为字符串 List&lt;String&gt; getAllRows(List&lt;List&lt;Object&gt;&gt; queryResult) {
 List&lt;String&gt; allRows = new ArrayList&lt;&gt;(); if (queryResult == null || queryResult.isEmpty()) return allRows; StringBuilder row = new StringBuilder(); for (List&lt;Object&gt; 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; } //更新offset元数据状态，每次返回结果集后调用。必须记录每次查询的offset值，为程序中断续跑数据时使用，以id为offset 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); } //执行sql语句 private void execSql(String sql) {
 try {
 ps = conn.prepareStatement(sql); LOG.info("exec::" + sql); ps.execute(); } catch (SQLException e) {
 e.printStackTrace(); } } //获取当前id的offset 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; } //查询一条数据的执行语句(当前id) private String queryOne(String sql) {
 ResultSet result = null; try {
 ps = conn.prepareStatement(sql); result = ps.executeQuery(); while (result.next()) {
 return result.getString(1); } } catch (SQLException e) {
 e.printStackTrace(); } return null; } //关闭相关资源 void close() {
 try {
 ps.close(); conn.close(); } catch (SQLException e) {
 e.printStackTrace(); } } int getCurrentIndex() {
 return currentIndex; } void setCurrentIndex(int newValue) {
 currentIndex = newValue; } int getRunQueryDelay() {
 return runQueryDelay; } String getQuery() {
 return query; } String getConnectionURL() {
 return connectionURL; } private boolean isCustomQuerySet() {
 return (customQuery != null); } Context getContext() {
 return context; } public String getConnectionUserName() {
 return connectionUserName; } public String getConnectionPassword() {
 return connectionPassword; } } </pre></td> 
 </tr> 
 </tbody> 
</table>

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

#### SQLSource ####

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

#### 测试准备 ####

#### 驱动包 ####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ cp \ /opt/sorfware/mysql-libs/mysql-connector-java-5.1.27/mysql-connector-java-5.1.27-bin.jar \ /opt/module/flume/lib/ </pre></td> 
 </tr> 
 </tbody> 
</table>

**打包项目并将jar放入flume的lib目录下**

#### 配置文件 ####

#### Mysql表准备 ####

#### 测试脚本 ####

#### 测试并查看 ####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/mysql.conf -Dflume.root.logger=INFO,console		#启动agent [hadoop@datanode1 job]$ sh mysql.sh #启动测试脚本 </pre></td> 
 </tr> 
 </tbody> 
</table>

![FqecK1.png][]

## Flume监控Ganglia ##

### 步骤 ###

1.**安装httpd服务与php**

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ sudo yum -y install httpd php </pre></td> 
 </tr> 
 </tbody> 
</table>

1.  安装其他依赖

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ sudo yum -y install rrdtool perl-rrdtool rrdtool-devel [hadoop@datanode1 flume]$ sudo yum -y install apr-devel </pre></td> 
 </tr> 
 </tbody> 
</table>

1. 安装ganglia
 
 <table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ sudo rpm -Uvh http://dl.fedoraproject.org/pub/epel/6/x86_64/epel-release-6-8.noarch.rpm [hadoop@datanode1 flume]$ sudo yum -y install ganglia-gmetad [hadoop@datanode1 flume]$ sudo yum -y install ganglia-web [hadoop@datanode1 flume]$ sudo yum install -y ganglia-gmond </pre></td> 
 </tr> 
 </tbody> 
 </table>
2. 修改ganglia

1.  修改配置文件gmetad.conf

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>data_source "datanode1" 192.168.1.101 </pre></td> 
 </tr> 
 </tbody> 
</table>

1.  修改配置文件gmond.conf

1.  修改配置文件config

注意selinux本次生效关闭必须重启，如果此时不想重启，可以临时生效之：

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ sudo setenforce 0 </pre></td> 
 </tr> 
 </tbody> 
</table>

### 启动ganglia ###

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ sudo service httpd start Starting httpd: [hadoop@datanode1 flume]$ sudo service gmetad start [hadoop@datanode1 flume]$ sudo service gmond start [hadoop@datanode1 flume]$ </pre></td> 
 </tr> 
 </tbody> 
</table>

如果完成以上操作依然出现权限不足错误，请修改/var/lib/ganglia目录的权限

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>[hadoop@datanode1 flume]$ sudo chmod -R 777 /var/lib/ganglia </pre></td> 
 </tr> 
 </tbody> 
</table>

### 操作Flume测试监控 ###

1.修改/opt/module/flume/conf目录下的flume-env.sh配置：

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 </pre></td> 
 <td><pre>[hadoop@datanode1 conf]$ vim flume-env.sh JAVA_OPTS="-Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.1.101:8649 -Xms100m -Xmx200m" [hadoop@datanode1 conf]$ xsync flume-env.sh </pre></td> 
 </tr> 
 </tbody> 
</table>

1.  启动flume任务

![FqnX9g.png][]

![FquPEV.png][]

<table> 
 <thead> 
 <tr> 
 <th>字段（图表名称）</th> 
 <th>字段含义</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>EventPutAttemptCount</td> 
 <td>source尝试写入channel的事件总数量</td> 
 </tr> 
 <tr> 
 <td>EventPutSuccessCount</td> 
 <td>成功写入channel且提交的事件总数量</td> 
 </tr> 
 <tr> 
 <td>EventTakeAttemptCount</td> 
 <td>sink尝试从channel拉取事件的总数量。这不意味着每次事件都被返回，因为sink拉取的时候channel可能没有任何数据。</td> 
 </tr> 
 <tr> 
 <td>EventTakeSuccessCount</td> 
 <td>sink成功读取的事件的总数量</td> 
 </tr> 
 <tr> 
 <td>StartTime</td> 
 <td>channel启动的时间（毫秒）</td> 
 </tr> 
 <tr> 
 <td>StopTime</td> 
 <td>channel停止的时间（毫秒）</td> 
 </tr> 
 <tr> 
 <td>ChannelSize</td> 
 <td>目前channel中事件的总数量</td> 
 </tr> 
 <tr> 
 <td>ChannelFillPercentage</td> 
 <td>channel占用百分比</td> 
 </tr> 
 <tr> 
 <td>ChannelCapacity</td> 
 <td>channel的容量</td> 
 </tr> 
 </tbody> 
</table>

[Fb1qot.png]: /images/20220210/81fe82cc207b426188b0ad0e81a1bee8.png
[Fb0ndU.png]: /images/20220210/d62706737aec4539940b40abb7a01cac.png
[Fb05Ss.png]: /images/20220210/00fb2aecef9646f5b1c41d005498f86a.png
[Fb2PFP.png]: /images/20220210/746a4e5223bf4b4cb25fd29fd9ad8985.png
[Fb2uoq.png]: /images/20220210/ce3c080fa0e44fd7ac9af0df9ad949b5.png
[FSNdhR.png]: /images/20220210/76daa1df63c54178aef6ae6430df43ad.png
[Fb2Gy4.png]: /images/20220210/cad2623473634854990b6bec98edba8c.png
[FSN9kd.png]: /images/20220210/c8c577ed14d24cedaea149ed1b9202f3.png
[FbzfSO.png]: /images/20220210/ae505c52f3fa4c6da6ed3b7c9c0ac2ae.png
[Fq9gkn.png]: /images/20220210/e27d45ee1a1047258e88cd169a9ca642.png
[FSN110.png]: /images/20220210/aafb3e72028f459ab5a71bd12b41c993.png
[FqFh59.png]: /images/20220210/1cbe3647ee27418080d5619eb656c810.png
[FqecK1.png]: /images/20220210/eb28dcde0f244fa58d2513f90fe19342.png
[FqnX9g.png]: /images/20220210/eb7e6f20b9934a5bb2bf7e4e9868323e.png
[FquPEV.png]: /images/20220210/24aeb064a0014edeb0dad141aaab6e33.png