云计算课程设计（Prometheus+grafana+Flume+ganglia+mysql+jdk）

一、准备情况

   prometheus下载所在: https://github.com/prometheus/prometheus/releases/download/v2.52.0-rc.1/prometheus-2.52.0-rc.1. windows-amd64.zip
grafana 下 载 地 址 :
  https://dl.grafana.com/enterprise/release/grafana-enterprise-10.4.2.windows-amd64.zip windows_exporter下载所在:
  https://github.com/prometheus-community/windows_exporter/releases/download/v0.25.1/wi ndows_exporter-0.25.1-amd64.msi
flume下载所在:
  https://archive.apache.org/dist/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz
jdk（linux）下载所在:
  https://cola-yunos-1305721388.cos.ap-guangzhou.myqcloud.com/20210813/jdk-8u401-linux-x64. tar.gz
mysql下载所在
https://downloads.mysql.com/archives/get/p/23/file/mysql-community-server-5.7.26-1.el7.x86_64.rpm
  https://downloads.mysql.com/archives/get/p/23/file/mysql-community-client-5.7.26-1.el7.x86_64.rpm
  https://downloads.mysql.com/archives/get/p/23/file/mysql-community-common-5.7.26-1.el7.x86_64.rpm
  https://downloads.mysql.com/archives/get/p/23/file/mysql-community-libs-5.7.26-1.el7.x86_64.rpm
驱动包下载所在:  
  https://downloads.mysql.com/archives/get/p/3/file/mysql-connector-java-5.1.16.tar.gz idea
  idea下载所在: 
https://www.jetbrains.com/zh-cn/idea/download/download-thanks.html?platform=windows 
maven下载所在: 
https://dlcdn.apache.org/maven/maven-3/3.8.8/binaries/apache-maven-3.8.8-bin.zip
  二、监督面板搭建

1、解压Prometheus，运行prometheus.exe

2、打开网址

3、配置windows服务器监督：

        双击运行windows_exporter-0.25.1-amd64.msi，在欣赏器进入以下所在：
http://10.225.193.16:9182/metrics，出现以下页面即为运行成功：

4、编辑prometheus.yml（此文件在解压的Prometheus内里）

在文件末段添加以下内容：

1. - job_name: "windows"
2.     file_sd_configs:
3.     - refresh_interval: 15s
4.       files:
5.       - ".\\windows.yml"

复制代码

5、在同级目录下新建windows.yml

1. - targets: ["127.0.0.1:9182"]
2.   labels:
3.     instance: 127.0.0.1
4. serverName: 'windows server'

复制代码

 

6、重启prometheus（就是把黑框框关掉，然后再双击prometheus.exe）

在欣赏器打开：http://127.0.0.1:9090/targets?search=

7、解压grafana，在bin目录打开grafana-server.exe 

8、打开网址

http://127.0.0.1:3000
默认的账号暗码都是admin 

9、添加datasource

 

出现下边页面即为搭建成功

三、Flume日记聚集搭建

1、创建目录

在‘/opt’目录下创建 software和module两个目录，命令如下：

1. mkdir module software

复制代码

2、利用xftp工具上传文件

3、将JDK和Flume文件解压到‘/opt/module’

1. tar -zxf jdk-8u371-linux-x64.tar.gz -C /opt/module/
2. tar -zxf apache-flume-1.6.0-bin.tar.gz -C /opt/module/

复制代码

4、修改安装目录名称

1. mv jdk1.8.0_371/ jdk
2. mv apache-flume-1.6.0-bin/ flume

复制代码

5、配置情况

A）JDK情况配置

[root@localhost module]# vi /etc/profile
配置内容如下：

1. export JAVA_HOME=/opt/module/jdk
2. export PATH=$PATH:$JAVA_HOME/bin # 将 JAVA 安装目录加入 PATH 路径

复制代码

B）Flume情况配置：

1. 配置Flume文件
2. [root@localhost conf]# mv flume-env.sh.template flume-env.sh

复制代码

1. # 编辑配置文件
2. [root@localhost conf]# vi flume-env.sh

复制代码

6、刷新情况变量

1. [root@localhost module]# source /etc/profile

复制代码

四、ganglia监控搭建

1、安装相关依靠包

1. [root@localhost ~]# yum -y install httpd php
3. [root@localhost ~]# yum -y install rrdtool perl-rrdtool rrdtool-devel
5. [root@localhost ~]# yum -y install apr-devel
7. [root@localhost ~]# yum install -y epel-release
9. [root@localhost ~]# yum -y install ganglia-gmetad
11. [root@localhost ~]# yum -y install ganglia-web
13. [root@localhost ~]# yum install -y ganglia-gmond

复制代码

2、安装telent

1. [root@localhost ~]# yum install telnet -y

复制代码

3、修改配置文件

1. [root@localhost ~]# vi /etc/httpd/conf.d/ganglia.conf

复制代码

1. [root@localhost ~]# vi /etc/ganglia/gmetad.conf

复制代码

1. 修改host文件
2. [root@localhost ~]# vi /etc/hosts

复制代码

1. [root@localhost ~]# vi /etc/ganglia/gmond.conf

复制代码

五、禁用selinux

1. [root@localhost ~]# vi /etc/selinux/config

复制代码

1、设置服务自启动

1. [root@master ~]# systemctl enable httpd && systemctl enable gmetad && systemctl enable gmond

复制代码

2、关闭防火墙

1. [root@master ~]# systemctl stop firewalld && systemctl disable firewalld

复制代码

3、授权

1. [root@master ~]# chmod -R 777 /var/lib/ganglia

复制代码

4、重启捏造机

1. [root@master ~]# reboot

复制代码

5、访问ganglia

http://192.168.229.160/ganglia/

6、修改flume配置文件

[root@master ~]# vi /opt/module/flume/conf/flume-env.sh
添加以下内容：

1. JAVA_OPTS="-Dflume.monitoring.type=ganglia
2. -Dflume.monitoring.hosts=192.168.229.160:8649  #（修改IP）
3. -Xms100m
4. -Xmx200m"

复制代码

7、创建job文件

在‘/opt/module/flume’目录下创建文件夹‘job’

进入该目录下创建一个名字为‘flume-telnet-logger.conf’文件，并编辑该文件，编辑内容如下：
[root@master job]# vi flume-telnet-logger.conf

1. # Name the components on this agent
2. a1.sources = r1
3. a1.sinks = k1
4. a1.channels = c1
6. # Describe/configure the source
7. a1.sources.r1.type = netcat
8. a1.sources.r1.bind = 0.0.0.0
9. a1.sources.r1.port = 44444
11. # Describe the sink
12. a1.sinks.k1.type = logger
14. # Use a channel which buffers events in memory
15. a1.channels.c1.type = memory
16. a1.channels.c1.capacity = 1000
17. a1.channels.c1.transactionCapacity = 100
19. # Bind the source and sink to the channel
20. a1.sources.r1.channels = c1
21. a1.sinks.k1.channel = c1

复制代码

8、启动Flume服务

1. [root@master ~]# /opt/module/flume/bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-telnet-logger.conf -Dflume.root.logger=INFO,console -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.229.160:8649

复制代码

9、重新开启一个xshell端口，发送数据

1. [root@master ~]# telnet localhost 44444

复制代码

 

10、查察网页端监控情况

http://192.168.229.160/ganglia/?r=hour&cs=&ce=&c=master&h=master&tab=m&vn=&tz=&hide-hf=false&m=load_one&sh=1&z=small&hc=4&host_regex=&max_graphs=0&s=by+name
 

六、flume日记聚集(MySQL)

1、利用xftp工具上传文件

2、删除mariadb依靠

系统中已经安装了 mariadb-libs-5.5.52-2.el7.x86_64 软件包，需要将其卸载

1. [root@master ~]# rpm -e mariadb-libs-5.5.65-1.el7.x86_64 --nodeps

复制代码

已删除。
3、安装mysql服务

1. [root@master software]# rpm -ivh mysql-community-common-5.7.18-1.el7.x86_64.rpm --nodeps
3. [root@master software]# rpm -ivh mysql-community-libs-5.7.18-1.el7.x86_64.rpm --nodeps
5. [root@master software]# rpm -ivh mysql-community-client-5.7.18-1.el7.x86_64.rpm --nodeps
7. [root@master software]# rpm -ivh mysql-community-server-5.7.18-1.el7.x86_64.rpm --nodeps

复制代码

4、启动MySQL服务

1. [root@master software]# systemctl start mysqld

复制代码

已启动。
5、查找MySQL初始暗码

1. [root@master software]# grep "password" /var/log/mysqld.log

复制代码

6、登录MySQL

1. [root@master ~]# mysql -uroot -p
3. # 输入初始密码：>q.r#A9(f&4<

复制代码

7、设置暗码复杂度

1. set global validate_password_policy=LOW;

复制代码

1. set global validate_password_length=4;

复制代码

8、修改暗码, 这里设置为root

1. alter user 'root'@'localhost' identified by 'root';

复制代码

9、开启远程访问

1. GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY 'root' WITH GRANT OPTION;

复制代码

10、利用navicat工具远程连接MySQL

确保MySQL服务正在运行，利用navicat举行远程访问：

连接成功
七、创建msyql source jar包

1、安装maven，在windows本机解压后打开conf目录下的setting.conf，添加阿里云镜像源

2、打开idea，新建一个maven项目

项目具体信息，如果本机没有jdk需要安装jdk

3、导入依靠

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
7.     <groupId>org.example</groupId>
8.     <artifactId>mysql_sourcde</artifactId>
9.     <version>1.0-SNAPSHOT</version>
11.     <properties>
12.         <maven.compiler.source>8</maven.compiler.source>
13.         <maven.compiler.target>8</maven.compiler.target>
14.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
15.     </properties>
16.     <dependencies>
17.         <dependency>
18.             <groupId>org.apache.flume</groupId>
19.             <artifactId>flume-ng-core</artifactId>
20.             <version>1.6.0</version>
21.         </dependency>
22.         <dependency>
23.             <groupId>mysql</groupId>
24.             <artifactId>mysql-connector-java</artifactId>
25.             <version>5.1.16</version>
26.         </dependency>
27.         <dependency>
28.             <groupId>log4j</groupId>
29.             <artifactId>log4j</artifactId>
30.             <version>1.2.17</version>
31.         </dependency>
32.         <dependency>
33.             <groupId>org.slf4j</groupId>
34.             <artifactId>slf4j-api</artifactId>
35.             <version>1.7.12</version>
36.         </dependency>
37.         <dependency>
38.             <groupId>org.slf4j</groupId>
39.             <artifactId>slf4j-log4j12</artifactId>
40.             <version>1.7.12</version>
41.         </dependency>
42.     </dependencies>
45. </project>

复制代码

刷新项目

4、创建目录

5、创建资源文件夹

6、创建两个资源文件jdbc.properties和log4j.properties

7、添加配置

jdbc.properties配置文件如下：

1. dbDriver=com.mysql.jdbc.Driver
2. dbUrl=jdbc:mysql://192.168.229.160:3306/mysqlsource? useUnicode=true&characterEncoding=utf-8
3. dbUser=root
4. dbPassword=root

复制代码

log4j.properties配置文件如下：

1. #--------console-----------
2. 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
4. #log4j.rootLogger=error,myfile
5. 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

复制代码

8、右键java目录，创建包

9、在包内新建两个类

SQLSource代码：

1. package com.nuit.source;
2. import org.apache.flume.Context;
3. import org.apache.flume.Event;
4. import org.apache.flume.EventDeliveryException;
5. import org.apache.flume.PollableSource;
6. import org.apache.flume.conf.Configurable;
7. import org.apache.flume.event.SimpleEvent;
8. import org.apache.flume.source.AbstractSource;
9. import org.slf4j.Logger;
10. import org.slf4j.LoggerFactory;
12. import java.text.ParseException;
13. import java.util.ArrayList;
14. import java.util.HashMap;
15. import java.util.List;
17. public class SQLSource extends AbstractSource implements Configurable, PollableSource {
19.     //打印日志
20.     private static final Logger LOG = LoggerFactory.getLogger(SQLSource.class);
22.     //定义sqlHelper
23.     private SQLSourceHelper sqlSourceHelper;
26.     @Override
27.     public long getBackOffSleepIncrement() {
28.         return 0;
29.     }
31.     @Override
32.     public long getMaxBackOffSleepInterval() {
33.         return 0;
34.     }
36.     @Override
37.     public void configure(Context context) {
39.         try {
40.             //初始化
41.             sqlSourceHelper = new SQLSourceHelper(context);
42.         } catch (ParseException e) {
43.             e.printStackTrace();
44.         }
45.     }
47.     @Override
48.     public Status process() throws EventDeliveryException {
50.         try {
51.             //查询数据表
52.             List<List<Object>> result = sqlSourceHelper.executeQuery();
54.             //存放event的集合
55.             List<Event> events = new ArrayList<>();
57.             //存放event头集合
58.             HashMap<String, String> header = new HashMap<>();
60.             //如果有返回数据，则将数据封装为event
61.             if (!result.isEmpty()) {
63.                 List<String> allRows = sqlSourceHelper.getAllRows(result);
65.                 Event event = null;
67.                 for (String row : allRows) {
68.                     event = new SimpleEvent();
69.                     event.setBody(row.getBytes());
70.                     event.setHeaders(header);
71.                     events.add(event);
72.                 }
74.                 //将event写入channel
75.                 this.getChannelProcessor().processEventBatch(events);
77.                 //更新数据表中的offset信息
78.                 sqlSourceHelper.updateOffset2DB(result.size());
79.             }
81.             //等待时长
82.             Thread.sleep(sqlSourceHelper.getRunQueryDelay());
84.             return Status.READY;
85.         } catch (InterruptedException e) {
86.             LOG.error("Error procesing row", e);
88.             return Status.BACKOFF;
89.         }
90.     }
92.     @Override
93.     public synchronized void stop() {
95.         LOG.info("Stopping sql source {} ...", getName());
97.         try {
98.             //关闭资源
99.             sqlSourceHelper.close();
100.         } finally {
101.             super.stop();
102.         }
103.     }
104. }

复制代码

SQLSourceHelper代码：

1. package com.nuit.source;
2. import org.apache.flume.Context;
3. import org.apache.flume.conf.ConfigurationException;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
7. import java.io.IOException;
8. import java.sql.*;
9. import java.text.ParseException;
10. import java.util.ArrayList;
11. import java.util.List;
12. import java.util.Properties;
14. public class SQLSourceHelper {
16.     private static final Logger LOG = LoggerFactory.getLogger(SQLSourceHelper.class);
18.     private int runQueryDelay, //两次查询的时间间隔
19.             startFrom,            //开始id
20.             currentIndex,                //当前id
21.             recordSixe = 0,      //每次查询返回结果的条数
22.             maxRow;                //每次查询的最大条数
24.     private String table,       //要操作的表
25.             columnsToSelect,     //用户传入的查询的列
26.             customQuery,          //用户传入的查询语句
27.             query,                 //构建的查询语句
28.             defaultCharsetResultSet;//编码集
30.     //上下文，用来获取配置文件
31.     private Context context;
33.     //为定义的变量赋值（默认值），可在flume任务的配置文件中修改
34.     private static final int DEFAULT_QUERY_DELAY = 10000;
35.     private static final int DEFAULT_START_VALUE = 0;
36.     private static final int DEFAULT_MAX_ROWS = 2000;
37.     private static final String DEFAULT_COLUMNS_SELECT = "*";
38.     private static final String DEFAULT_CHARSET_RESULTSET = "UTF-8";
40.     private static Connection conn = null;
41.     private static PreparedStatement ps = null;
42.     private static String connectionURL, connectionUserName, connectionPassword;
44.     //加载静态资源
45.     static {
47.         Properties p = new Properties();
49.         try {
50.             p.load(SQLSourceHelper.class.getClassLoader().getResourceAsStream("jdbc.properties"));
51.             connectionURL = p.getProperty("dbUrl");
52.             connectionUserName = p.getProperty("dbUser");
53.             connectionPassword = p.getProperty("dbPassword");
54.             Class.forName(p.getProperty("dbDriver"));
56.         } catch (IOException | ClassNotFoundException e) {
57.             LOG.error(e.toString());
58.         }
59.     }
61.     //获取JDBC连接
62.     private static Connection InitConnection(String url, String user, String pw) {
63.         try {
65.             Connection conn = DriverManager.getConnection(url, user, pw);
67.             if (conn == null)
68.                 throw new SQLException();
70.             return conn;
72.         } catch (SQLException e) {
73.             e.printStackTrace();
74.         }
76.         return null;
77.     }
79.     //构造方法
80.     SQLSourceHelper(Context context) throws ParseException {
82.         //初始化上下文
83.         this.context = context;
85.         //有默认值参数：获取flume任务配置文件中的参数，读不到的采用默认值
86.         this.columnsToSelect = context.getString("columns.to.select", DEFAULT_COLUMNS_SELECT);
88.         this.runQueryDelay = context.getInteger("run.query.delay", DEFAULT_QUERY_DELAY);
90.         this.startFrom = context.getInteger("start.from", DEFAULT_START_VALUE);
92.         this.defaultCharsetResultSet = context.getString("default.charset.resultset", DEFAULT_CHARSET_RESULTSET);
94.         //无默认值参数：获取flume任务配置文件中的参数
95.         this.table = context.getString("table");
96.         this.customQuery = context.getString("custom.query");
98.         connectionURL = context.getString("connection.url");
100.         connectionUserName = context.getString("connection.user");
102.         connectionPassword = context.getString("connection.password");
104.         conn = InitConnection(connectionURL, connectionUserName, connectionPassword);
106.         //校验相应的配置信息，如果没有默认值的参数也没赋值，抛出异常
107.         checkMandatoryProperties();
109.         //获取当前的id
110.         currentIndex = getStatusDBIndex(startFrom);
112.         //构建查询语句
113.         query = buildQuery();
114.     }
116.     //校验相应的配置信息（表，查询语句以及数据库连接的参数）
117.     private void checkMandatoryProperties() {
119.         if (table == null) {
120.             throw new ConfigurationException("property table not set");
121.         }
123.         if (connectionURL == null) {
124.             throw new ConfigurationException("connection.url property not set");
125.         }
127.         if (connectionUserName == null) {
128.             throw new ConfigurationException("connection.user property not set");
129.         }
131.         if (connectionPassword == null) {
132.             throw new ConfigurationException("connection.password property not set");
133.         }
134.     }
136.     //构建sql语句
137.     private String buildQuery() {
139.         String sql = "";
141.         //获取当前id
142.         currentIndex = getStatusDBIndex(startFrom);
143.         LOG.info(currentIndex + "");
145.         if (customQuery == null) {
146.             sql = "SELECT " + columnsToSelect + " FROM " + table;
147.         } else {
148.             sql = customQuery;
149.         }
151.         StringBuilder execSql = new StringBuilder(sql);
153.         //以id作为offset
154.         if (!sql.contains("where")) {
155.             execSql.append(" where ");
156.             execSql.append("id").append(">").append(currentIndex);
158.             return execSql.toString();
159.         } else {
160.             int length = execSql.toString().length();
162.             return execSql.toString().substring(0, length - String.valueOf(currentIndex).length()) + currentIndex;
163.         }
164.     }
166.     //执行查询
167.     List<List<Object>> executeQuery() {
169.         try {
170.             //每次执行查询时都要重新生成sql，因为id不同
171.             customQuery = buildQuery();
173.             //存放结果的集合
174.             List<List<Object>> results = new ArrayList<>();
176.             if (ps == null) {
177.                 //
178.                 ps = conn.prepareStatement(customQuery);
179.             }
181.             ResultSet result = ps.executeQuery(customQuery);
183.             while (result.next()) {
185.                 //存放一条数据的集合（多个列）
186.                 List<Object> row = new ArrayList<>();
188.                 //将返回结果放入集合
189.                 for (int i = 1; i <= result.getMetaData().getColumnCount(); i++) {
190.                     row.add(result.getObject(i));
191.                 }
193.                 results.add(row);
194.             }
196.             LOG.info("execSql:" + customQuery + "\nresultSize:" + results.size());
198.             return results;
199.         } catch (SQLException e) {
200.             LOG.error(e.toString());
202.             // 重新连接
203.             conn = InitConnection(connectionURL, connectionUserName, connectionPassword);
205.         }
207.         return null;
208.     }
210.     //将结果集转化为字符串，每一条数据是一个list集合，将每一个小的list集合转化为字符串
211.     List<String> getAllRows(List<List<Object>> queryResult) {
213.         List<String> allRows = new ArrayList<>();
215.         if (queryResult == null || queryResult.isEmpty())
216.             return allRows;
218.         StringBuilder row = new StringBuilder();
220.         for (List<Object> rawRow : queryResult) {
222.             Object value = null;
224.             for (Object aRawRow : rawRow) {
226.                 value = aRawRow;
228.                 if (value == null) {
229.                     row.append(",");
230.                 } else {
231.                     row.append(aRawRow.toString()).append(",");
232.                 }
233.             }
235.             allRows.add(row.toString());
236.             row = new StringBuilder();
237.         }
239.         return allRows;
240.     }
242.     //更新offset元数据状态，每次返回结果集后调用。必须记录每次查询的offset值，为程序中断续跑数据时使用，以id为offset
243.     void updateOffset2DB(int size) {
244.         //以source_tab做为KEY，如果不存在则插入，存在则更新（每个源表对应一条记录）
245.         String sql = "insert into flume_meta(source_tab,currentIndex) VALUES('"
246.                 + this.table
247.                 + "','" + (recordSixe += size)
248.                 + "') on DUPLICATE key update source_tab=values(source_tab),currentIndex=values(currentIndex)";
250.         LOG.info("updateStatus Sql:" + sql);
252.         execSql(sql);
253.     }
255.     //执行sql语句
256.     private void execSql(String sql) {
258.         try {
259.             ps = conn.prepareStatement(sql);
261.             LOG.info("exec::" + sql);
263.             ps.execute();
264.         } catch (SQLException e) {
265.             e.printStackTrace();
266.         }
267.     }
269.     //获取当前id的offset
270.     private Integer getStatusDBIndex(int startFrom) {
272.         //从flume_meta表中查询出当前的id是多少
273.         String dbIndex = queryOne("select currentIndex from flume_meta where source_tab='" + table + "'");
275.         if (dbIndex != null) {
276.             return Integer.parseInt(dbIndex);
277.         }
279.         //如果没有数据，则说明是第一次查询或者数据表中还没有存入数据，返回最初传入的值
280.         return startFrom;
281.     }
283.     //查询一条数据的执行语句(当前id)
284.     private String queryOne(String sql) {
286.         ResultSet result = null;
288.         try {
289.             ps = conn.prepareStatement(sql);
290.             result = ps.executeQuery();
292.             while (result.next()) {
293.                 return result.getString(1);
294.             }
295.         } catch (SQLException e) {
296.             e.printStackTrace();
297.         }
299.         return null;
300.     }
302.     //关闭相关资源
303.     void close() {
305.         try {
306.             ps.close();
307.             conn.close();
308.         } catch (SQLException e) {
309.             e.printStackTrace();
310.         }
311.     }
313.     int getCurrentIndex() {
314.         return currentIndex;
315.     }
317.     void setCurrentIndex(int newValue) {
318.         currentIndex = newValue;
319.     }
321.     int getRunQueryDelay() {
322.         return runQueryDelay;
323.     }
325.     String getQuery() {
326.         return query;
327.     }
329.     String getConnectionURL() {
330.         return connectionURL;
331.     }
333.     private boolean isCustomQuerySet() {
334.         return (customQuery != null);
335.     }
337.     Context getContext() {
338.         return context;
339.     }
341.     public String getConnectionUserName() {
342.         return connectionUserName;
343.     }
345.     public String getConnectionPassword() {
346.         return connectionPassword;
347.     }
349.     String getDefaultCharsetResultSet() {
350.         return defaultCharsetResultSet;
351.     }
352. }

复制代码

10、打jar包

打包结果：

11、打包成功后，target目录会出现一个jar包

12、打开jar包的位置，并利用xftp工具上传到服务器

13、添加jar包依靠

1. # 解压mysql-connector-java-5.1.16.tar.gz
2. [root@master software]# tar -zxf mysql-connector-java-5.1.16.tar.gz -C /opt/module/
4. # 复制‘mysql-connector-java-5.1.16/mysql-connector-java-5.1.16-bin.jar’、‘mysql_source-1.0-SNAPSHOT.jar’两个压缩包到‘/opt/module/flume/lib/‘目录下
6. [root@master software]# cp -p mysql_source-1.0-SNAPSHOT.jar /opt/module/flume/lib/
8. [root@master module]# cp -p mysql-connector-java-5.1.16/mysql-connector-java-5.1.16-bin.jar /opt/module/flume/lib/

复制代码

14、创建job文件

[root@master job]# pwd
/opt/module/flume/job
[root@master job]# vi mysql.conf
编辑内容如下：

1. # Name the components on this agent
2. a1.sources = r1
3. a1.sinks = k1
4. a1.channels = c1
6. # Describe/configure the source
7. a1.sources.r1.type = com.nuit.source.SQLSource  
8. a1.sources.r1.connection.url = jdbc:mysql://192.168.229.160:3306/mysqlsource
9. a1.sources.r1.connection.user = root
10. a1.sources.r1.connection.password = root
11. a1.sources.r1.table = student
12. a1.sources.r1.columns.to.select = *
13. #a1.sources.r1.incremental.column.name = id
14. #a1.sources.r1.incremental.value = 0
15. a1.sources.r1.run.query.delay=5000
17. # Describe the sink
18. a1.sinks.k1.type = logger
20. # Describe the channel
21. a1.channels.c1.type = memory
22. a1.channels.c1.capacity = 1000
23. a1.channels.c1.transactionCapacity = 100
25. # Bind the source and sink to the channel
26. a1.sources.r1.channels = c1
27. a1.sinks.k1.channel = c1

复制代码

八、测试结果

1、创建一个名为’mysqlsource‘数据库

2、选择刚刚创建的数据库

3、创建数据表

1. CREATE TABLE `student` (
2. `id` int(11) NOT NULL AUTO_INCREMENT,
3. `name` varchar(255) NOT NULL,
4. PRIMARY KEY (`id`)
5. );
6. CREATE TABLE `flume_meta` (
7. `source_tab` varchar(255) NOT NULL,
8. `currentIndex` varchar(255) NOT NULL,
9. PRIMARY KEY (`source_tab`)
10. );

复制代码

4、向表中插入数据

1. insert into student values(1,'zhangsan');
2. insert into student values(2,'lisi');
3. insert into student values(3,'wangwu');
4. insert into student values(4,'zhaoliu');

复制代码

5、查察表的数据

6、退出MySQL，启动flume服务

1. [root@master flume]# bin/flume-ng agent --conf conf/ --name a1 \
2. > --conf-file job/mysql.conf -Dflume.root.logger=INFO,console

复制代码

发现我们插入数据的日记，至此搭建完成！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。