云计算课程设计（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
https://i-blog.csdnimg.cn/direct/456c838c2301422f87ee1a703f6a6169.png
2、打开网址
https://i-blog.csdnimg.cn/direct/d6c8eed6e1ba479d99e5b41490680254.png
3、配置windows服务器监督：
双击运行windows_exporter-0.25.1-amd64.msi，在欣赏器进入以下所在：
http://10.225.193.16:9182/metrics，出现以下页面即为运行成功：
https://i-blog.csdnimg.cn/direct/b824d083bab24de39f4bf29a24f5226c.png
4、编辑prometheus.yml（此文件在解压的Prometheus内里）
在文件末段添加以下内容：
- job_name: "windows"
file_sd_configs:
- refresh_interval: 15s
files:
- ".\\windows.yml" https://i-blog.csdnimg.cn/direct/0edf5f6e58364b7f9552e0f4643cd168.png
5、在同级目录下新建windows.yml
- targets: ["127.0.0.1:9182"]
labels:
instance: 127.0.0.1
serverName: 'windows server' https://i-blog.csdnimg.cn/direct/3c75d093957444a29a1269c82d97d5ea.png
https://i-blog.csdnimg.cn/direct/d79984b2afa54984bb5c9e74ea91a203.png
6、重启prometheus（就是把黑框框关掉，然后再双击prometheus.exe）
在欣赏器打开：http://127.0.0.1:9090/targets?search=
https://i-blog.csdnimg.cn/direct/c0a0e06cdcb24133aa82ca89e4bc4ad4.png
7、解压grafana，在bin目录打开grafana-server.exe
https://i-blog.csdnimg.cn/direct/019ff74cf8c84d9d805c78d119dd09be.png
8、打开网址
http://127.0.0.1:3000
默认的账号暗码都是admin
https://i-blog.csdnimg.cn/direct/d961138a8cbc4c9aac7d583ae2c92162.png
9、添加datasource
https://i-blog.csdnimg.cn/direct/f3e0a3ff1b0b4f18ad6fd687b36dcd71.png
https://i-blog.csdnimg.cn/direct/d0b356ac41614e1d96c8e27b20361542.png
https://i-blog.csdnimg.cn/direct/088daf3f5aac421b90df240676694272.png
https://i-blog.csdnimg.cn/direct/dec4fdf4888446748adb7344c4b0797f.png
https://i-blog.csdnimg.cn/direct/976f3f979742441a979c0ea8d62709be.png
https://i-blog.csdnimg.cn/direct/dda9944fd9dc456184f152fe88580fd8.png
出现下边页面即为搭建成功
https://i-blog.csdnimg.cn/direct/7cc5a12014e44ad4ba4c83e9579f4b6a.png
三、Flume日记聚集搭建
1、创建目录
在‘/opt’目录下创建 software和module两个目录，命令如下：
mkdir module software https://i-blog.csdnimg.cn/direct/3947919b0ba842af8f1645f6fc59f5ba.png
2、利用xftp工具上传文件
https://i-blog.csdnimg.cn/direct/cdbf1634e1d644f481a00831e3e0d6ba.png
3、将JDK和Flume文件解压到‘/opt/module’
tar -zxf jdk-8u371-linux-x64.tar.gz -C /opt/module/
tar -zxf apache-flume-1.6.0-bin.tar.gz -C /opt/module/ 4、修改安装目录名称
mv jdk1.8.0_371/ jdk
mv apache-flume-1.6.0-bin/ flume https://i-blog.csdnimg.cn/direct/f8e230c7375e4a558ac23733384dc150.png
5、配置情况
A）JDK情况配置
# vi /etc/profile
配置内容如下：
export JAVA_HOME=/opt/module/jdk
export PATH=$PATH:$JAVA_HOME/bin # 将 JAVA 安装目录加入 PATH 路径 https://i-blog.csdnimg.cn/direct/7e70611af4054a7985e3752726b8220d.png
B）Flume情况配置：
配置Flume文件
# mv flume-env.sh.template flume-env.sh https://i-blog.csdnimg.cn/direct/0051e61d457c4643872a50bc93eab2e6.png
# 编辑配置文件
# vi flume-env.sh https://i-blog.csdnimg.cn/direct/52932b336c8b42cbaa130245da04484c.png
6、刷新情况变量
# source /etc/profile https://i-blog.csdnimg.cn/direct/afe123c936934da28fa55a6a7edb32f4.png
四、ganglia监控搭建
1、安装相关依靠包
# yum -y install httpd php
# yum -y install rrdtool perl-rrdtool rrdtool-devel
# yum -y install apr-devel
# yum install -y epel-release
# yum -y install ganglia-gmetad
# yum -y install ganglia-web
# yum install -y ganglia-gmond 2、安装telent
# yum install telnet -y https://i-blog.csdnimg.cn/direct/5bb4c03f7aae470b9802c928a531a46e.png
3、修改配置文件
# vi /etc/httpd/conf.d/ganglia.conf https://i-blog.csdnimg.cn/direct/10e8f02ff6c54722ae33d973081b87a6.png
# vi /etc/ganglia/gmetad.conf https://i-blog.csdnimg.cn/direct/0cdbb2428462497bbcce207771557910.png
修改host文件
# vi /etc/hosts https://i-blog.csdnimg.cn/direct/ea536960d1db4fa099016f5d0fb849c0.png
# vi /etc/ganglia/gmond.conf https://i-blog.csdnimg.cn/direct/355255bcd4194450b5442ee50eb1072d.png
https://i-blog.csdnimg.cn/direct/e6ee4f95f7174bf6afc09602cccb4136.png
https://i-blog.csdnimg.cn/direct/ea6f8590825c4b38a2de5bd7b1b3d60e.png
五、禁用selinux
# vi /etc/selinux/config https://i-blog.csdnimg.cn/direct/65d1ff965a8b4fec999d3fd570888ce6.png
1、设置服务自启动
# systemctl enable httpd && systemctl enable gmetad && systemctl enable gmond https://i-blog.csdnimg.cn/direct/4a2e047952034970885fa48d6aa5cff3.png
2、关闭防火墙
# systemctl stop firewalld && systemctl disable firewalld 3、授权
# chmod -R 777 /var/lib/ganglia https://i-blog.csdnimg.cn/direct/040b217a6b2540e4bd0b40f7a3217098.png
4、重启捏造机
# reboot 5、访问ganglia
http://192.168.229.160/ganglia/
https://i-blog.csdnimg.cn/direct/f061086b52ff4a208a6ea820d0ec5e54.png
6、修改flume配置文件
# vi /opt/module/flume/conf/flume-env.sh
添加以下内容：
JAVA_OPTS="-Dflume.monitoring.type=ganglia
-Dflume.monitoring.hosts=192.168.229.160:8649#（修改IP）
-Xms100m
-Xmx200m" https://i-blog.csdnimg.cn/direct/305c53d6a84845bab86fb0da23d9ba68.png
7、创建job文件
在‘/opt/module/flume’目录下创建文件夹‘job’
https://i-blog.csdnimg.cn/direct/10e36d9b41c54f6b89617b6ac7002368.png
进入该目录下创建一个名字为‘flume-telnet-logger.conf’文件，并编辑该文件，编辑内容如下：
# vi flume-telnet-logger.conf
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1 8、启动Flume服务
# /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 https://i-blog.csdnimg.cn/direct/d9fa0a5e53874fa1b073c08abccc8575.png
9、重新开启一个xshell端口，发送数据
# telnet localhost 44444 https://i-blog.csdnimg.cn/direct/671d3f9a25b64c159ad3d73b4b5b89c3.png
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
https://i-blog.csdnimg.cn/direct/7ab53997dfec48fe91236b880f1acbb7.png
六、flume日记聚集(MySQL)
1、利用xftp工具上传文件
https://i-blog.csdnimg.cn/direct/97a1199f6bda4f75a92ade3849e38dc6.png
2、删除mariadb依靠
https://i-blog.csdnimg.cn/direct/7c4ef7d857784e35affe58726a3d4347.png
系统中已经安装了 mariadb-libs-5.5.52-2.el7.x86_64 软件包，需要将其卸载
# rpm -e mariadb-libs-5.5.65-1.el7.x86_64 --nodeps https://i-blog.csdnimg.cn/direct/4fc2746173ff487cb97d9eb2818778d2.png
已删除。
3、安装mysql服务
# rpm -ivh mysql-community-common-5.7.18-1.el7.x86_64.rpm --nodeps
# rpm -ivh mysql-community-libs-5.7.18-1.el7.x86_64.rpm --nodeps
# rpm -ivh mysql-community-client-5.7.18-1.el7.x86_64.rpm --nodeps
# rpm -ivh mysql-community-server-5.7.18-1.el7.x86_64.rpm --nodeps https://i-blog.csdnimg.cn/direct/6dd2483ffb924321a4d3612b13c045a1.png
4、启动MySQL服务
# systemctl start mysqld https://i-blog.csdnimg.cn/direct/75607350ffa342c5ae63a83f098099c7.png
已启动。
5、查找MySQL初始暗码
# grep "password" /var/log/mysqld.log https://i-blog.csdnimg.cn/direct/c09e2fee82f149219f1a7ace5001f944.png
6、登录MySQL
# mysql -uroot -p
# 输入初始密码：>q.r#A9(f&4< https://i-blog.csdnimg.cn/direct/6ca781edea304aa79b4a7bfa1ad040a7.png
7、设置暗码复杂度
set global validate_password_policy=LOW; https://i-blog.csdnimg.cn/direct/28049c0a173d4b73aa1a3348b3be8617.png
set global validate_password_length=4; https://i-blog.csdnimg.cn/direct/8432e78e4fd84f9e84bd2381cf82548a.png
8、修改暗码, 这里设置为root
alter user 'root'@'localhost' identified by 'root'; https://i-blog.csdnimg.cn/direct/5d9e8e6210814e2eb17f03676a8f49f5.png
9、开启远程访问
GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY 'root' WITH GRANT OPTION; https://i-blog.csdnimg.cn/direct/518768571b6843e69b6322a892670e27.png
10、利用navicat工具远程连接MySQL
确保MySQL服务正在运行，利用navicat举行远程访问：
https://i-blog.csdnimg.cn/direct/0e10a0176dac4d9f812e3d96bc54c488.png
https://i-blog.csdnimg.cn/direct/614e45fc25bd4aa98bda24b9156b673a.png连接成功
七、创建msyql source jar包
1、安装maven，在windows本机解压后打开conf目录下的setting.conf，添加阿里云镜像源
https://i-blog.csdnimg.cn/direct/345ced2785e84e3c875e07a5f93fe5cb.png
2、打开idea，新建一个maven项目
https://i-blog.csdnimg.cn/direct/5b3487715f2a4a4f8ff99e92268ba8e2.png
项目具体信息，如果本机没有jdk需要安装jdk
https://i-blog.csdnimg.cn/direct/2ea5a9a458ad4bc182f16acd6044c794.png
https://i-blog.csdnimg.cn/direct/0a307dae584b489b9a8dba2e2f7971d3.png
https://i-blog.csdnimg.cn/direct/b7eb87d34a7f479d9e7c68c988efa93b.png
3、导入依靠
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>mysql_sourcde</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-core</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.16</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.12</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.12</version>
</dependency>
</dependencies>
</project> 刷新项目
https://i-blog.csdnimg.cn/direct/3fbe5693e4e144c6b77288c584180f89.png
4、创建目录
https://i-blog.csdnimg.cn/direct/ae56746ffe56489b89eb954966f1a8de.png
https://i-blog.csdnimg.cn/direct/19d33348f11543dcb3d41ab5e8179126.png
5、创建资源文件夹
https://i-blog.csdnimg.cn/direct/bbcfd53a818f40529626c0ffb6504278.png
https://i-blog.csdnimg.cn/direct/5b4f4c2681124250a29b1ec72a0bac51.png
6、创建两个资源文件jdbc.properties和log4j.properties
https://i-blog.csdnimg.cn/direct/0befa91a895a435c81fbeb242c983d9e.png
https://i-blog.csdnimg.cn/direct/e316efb202b24212b911acbcacaee1c6.png
7、添加配置
jdbc.properties配置文件如下：
dbDriver=com.mysql.jdbc.Driver
dbUrl=jdbc:mysql://192.168.229.160:3306/mysqlsource? useUnicode=true&characterEncoding=utf-8
dbUser=root
dbPassword=root 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 8、右键java目录，创建包
https://i-blog.csdnimg.cn/direct/7a6604defe2d4521b07b02645ab5a70c.png
9、在包内新建两个类
https://i-blog.csdnimg.cn/direct/01e7243ec10441079daa13c7079102a7.png
https://i-blog.csdnimg.cn/direct/42e835e23b85494a8f0fcac224739b52.png
SQLSource代码：
package com.nuit.source;
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.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
public class SQLSource extends AbstractSource implements Configurable, PollableSource {
//打印日志
private static final Logger LOG = LoggerFactory.getLogger(SQLSource.class);
//定义sqlHelper
private SQLSourceHelper sqlSourceHelper;
@Override
public long getBackOffSleepIncrement() {
return 0;
}
@Override
public long getMaxBackOffSleepInterval() {
return 0;
}
@Override
public void configure(Context context) {
try {
//初始化
sqlSourceHelper = new SQLSourceHelper(context);
} catch (ParseException e) {
e.printStackTrace();
}
}
@Override
public Status process() throws EventDeliveryException {
try {
//查询数据表
List<List<Object>> result = sqlSourceHelper.executeQuery();
//存放event的集合
List<Event> events = new ArrayList<>();
//存放event头集合
HashMap<String, String> header = new HashMap<>();
//如果有返回数据，则将数据封装为event
if (!result.isEmpty()) {
List<String> allRows = sqlSourceHelper.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信息
sqlSourceHelper.updateOffset2DB(result.size());
}
//等待时长
Thread.sleep(sqlSourceHelper.getRunQueryDelay());
return Status.READY;
} catch (InterruptedException e) {
LOG.error("Error procesing row", e);
return Status.BACKOFF;
}
}
@Override
public synchronized void stop() {
LOG.info("Stopping sql source {} ...", getName());
try {
//关闭资源
sqlSourceHelper.close();
} finally {
super.stop();
}
}
} SQLSourceHelper代码：
package com.nuit.source;
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(">").append(currentIndex);
return execSql.toString();
} else {
int length = execSql.toString().length();
return execSql.toString().substring(0, length - String.valueOf(currentIndex).length()) + currentIndex;
}
}
//执行查询
List<List<Object>> executeQuery() {
try {
//每次执行查询时都要重新生成sql，因为id不同
customQuery = buildQuery();
//存放结果的集合
List<List<Object>> results = new ArrayList<>();
if (ps == null) {
//
ps = conn.prepareStatement(customQuery);
}
ResultSet result = ps.executeQuery(customQuery);
while (result.next()) {
//存放一条数据的集合（多个列）
List<Object> row = new ArrayList<>();
//将返回结果放入集合
for (int i = 1; i <= 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<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;
}
//更新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;
}
String getDefaultCharsetResultSet() {
return defaultCharsetResultSet;
}
} 10、打jar包
https://i-blog.csdnimg.cn/direct/0ec8e2545cbe4b25ba5b480fcdbc5ca0.png
打包结果：
https://i-blog.csdnimg.cn/direct/2dca7e8f5c4f41518d0dab95b62c5d7e.png
11、打包成功后，target目录会出现一个jar包
https://i-blog.csdnimg.cn/direct/6824041bd6bc4346949ddf4a64e8c18f.png
12、打开jar包的位置，并利用xftp工具上传到服务器
https://i-blog.csdnimg.cn/direct/aa1efc7d200b421ab44851d8b3563a09.png
https://i-blog.csdnimg.cn/direct/c7fc6a89a5c4483c9226942312428bdf.png
13、添加jar包依靠
# 解压mysql-connector-java-5.1.16.tar.gz
# tar -zxf mysql-connector-java-5.1.16.tar.gz -C /opt/module/
# 复制‘mysql-connector-java-5.1.16/mysql-connector-java-5.1.16-bin.jar’、‘mysql_source-1.0-SNAPSHOT.jar’两个压缩包到‘/opt/module/flume/lib/‘目录下
# cp -p mysql_source-1.0-SNAPSHOT.jar /opt/module/flume/lib/
# cp -p mysql-connector-java-5.1.16/mysql-connector-java-5.1.16-bin.jar /opt/module/flume/lib/ 14、创建job文件
# pwd
/opt/module/flume/job
# vi 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.nuit.source.SQLSource
a1.sources.r1.connection.url = jdbc:mysql://192.168.229.160:3306/mysqlsource
a1.sources.r1.connection.user = root
a1.sources.r1.connection.password = root
a1.sources.r1.table = student
a1.sources.r1.columns.to.select = *
#a1.sources.r1.incremental.column.name = id
#a1.sources.r1.incremental.value = 0
a1.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 = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1 八、测试结果
1、创建一个名为’mysqlsource‘数据库
https://i-blog.csdnimg.cn/direct/efa036f15e524a8eb32f28d403017882.png
2、选择刚刚创建的数据库
https://i-blog.csdnimg.cn/direct/a045b7b5c25342459b3161fbcb162bb6.png
3、创建数据表
CREATE TABLE `student` (
`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`)
); https://i-blog.csdnimg.cn/direct/9bddbd5e48f24d6d8416accb9200c696.png
4、向表中插入数据
insert into student values(1,'zhangsan');
insert into student values(2,'lisi');
insert into student values(3,'wangwu');
insert into student values(4,'zhaoliu'); https://i-blog.csdnimg.cn/direct/09d2ff762286453699184f67e4d2b85b.png
5、查察表的数据
https://i-blog.csdnimg.cn/direct/4ec8aa7ee3eb4a1c823f702a3125f81f.png
6、退出MySQL，启动flume服务
# bin/flume-ng agent --conf conf/ --name a1 \
> --conf-file job/mysql.conf -Dflume.root.logger=INFO,console https://i-blog.csdnimg.cn/direct/ef6a94acfb6b4a42873430396c3185eb.png
发现我们插入数据的日记，至此搭建完成！
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]