1、UI界面结果展示

UI界面基于 layui - layout 构建

1.1、Shell控制台首页

1.2、HDFS 操作页面

1.3、SparkStreaming & Kafka 操作页面

1.4、MapReduce の WordCount 操作界面

1.5、HBase 操作界面

新建表：

查看详情：

添加/更新/修改 数据：

删除列族：

删除行键：

删除表：

2、准备操作：

2.1、本机环境

本机所需环境：

• java
• scala
• mysql
• maven

2.2、配置集群环境变量

在三台虚拟机的/etc/profile中添加：

export JAVA_HOME=/opt/modules/jdk1.8.0_161
export PATH=$PATH:$JAVA_HOME/bin

export HADOOP_HOME=/opt/modules/hadoop-2.8.2
export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

export ZOOKEEPER_HOME=/opt/modules/zookeeper-3.4.10
export PATH=$PATH:$ZOOKEEPER_HOME/bin

export HBASE_HOME=/opt/modules/hbase-1.2.6.1
export PATH=$PATH:$HBASE_HOME/bin

export HIVE_HOME=/opt/modules/apache-hive-2.3.8-bin
export PATH=$PATH:$HIVE_HOME/bin

export SQOOP_HOME=/opt/modules/sqoop-1.4.7
export PATH=$PATH:$SQOOP_HOME/bin

export FLUME_HOME=/opt/modules/apache-flume-1.8.0-bin
export PATH=$PATH:$FLUME_HOME/bin

export KAFKA_HOME=/opt/modules/kafka_2.11-2.0.0
export PATH=$PATH:$KAFKA_HOME/bin

export KAFKA_CONFIG_HOME=/opt/modules/kafka_2.11-2.0.0/config
export PATH=$PATH:$KAFKA_CONFIG_HOME/bin

export STORM_HOME=/opt/modules/apache-storm-1.1.0
export PATH=$PATH:$STORM_HOME/bin

export ELASTICSEARCH_HOME=/opt/modules/elasticsearch-7.12.0
export PATH=$PATH:$ELASTICSEARCH_HOME/bin

export KIBANA_HOME=/opt/modules/kibana-7.12.0-linux-x86_64
export PATH=$PATH:$KIBANA_HOME/bin

export ELASTICSEARCH_HEAD_HOME=/opt/modules/node-v10.9.0-linux-x64
export PATH=$PATH:$ELASTICSEARCH_HEAD_HOME/bin

2.3、集群一键启动/关闭脚本

集群启动脚本start-cluster.sh：

#! /bin/bash
function help() { 
echo -e "\033[32;1m
start_cluster.sh默认会启动 Zookeeper ，其余的Hadoop组件则需要手动在命令行后面添加参数启动 

例如：
	启动 YARN HA ：start_cluster.sh yarn
	启动 YARN HA 和 HBase HA ：start_cluster.sh yarn hbase
	...

可选Hadoop组件参数有：
	hdfs                       -- 启动 HDFS HA                -- 
	yarn                       -- 启动 YARN HA                  | 
	hbase                      -- 启动 HBase HA                 |
	hive                       -- 启动 Hive                      >  经过实测，这些组件可以同时启动
	spark                      -- 启动 Spark                    | 
	kafka                      -- 启动 Kafka                    |
	storm                      -- 启动 Storm                  -- 
		*	*	*	*	*	*	*	*	*	
	elasticsearch              -- 启动 elasticsearch          --
	kibana	                   -- elasticsearch 是必填项	     >  这三个组件可以同时启动
	elasticsearch_head         -- elasticsearch 是必填项      --
		*	*	*	*	*	*	*	*	*
	all                        -- 一键启动全部组件            ！！！慎用 

    关于flume：本脚本不提供flume的一键启动，请使用命令
	-- flume-ng agent --conf c0nf --conf-file $FLUME_HOME/conf/flume-conf.properties --name a1 -Dflume.root.logger=INFO,console
    来启动flume监听44444端口，通过ssh远程登陆centos01，使用【telnet localhost 44444】命令来测试flume
\033[0m       \033[33;1m
一则警告 WARNING ；
	上面的这些组件如果想要全部同时启动，即执行
	    -- start_cluster.sh yarn hbase hive spark kafka storm elasticsearch kibana elasticserch_head
	    或
	    -- start_cluster.sh all
	命令，请确保你的虚拟机集群每台都至少有 6GB 的运行内存，否则后面的几个组件是起不来的，尤其是 elasticsearch 及其扩展组件
	极端情况下可能会导致虚拟机卡死甚至外部主机也直接卡死！！！\033[0m

"    
}

function start_zookeeper(){
echo -e "\033[32;1m
/****************************************************************/
/*                    启动 ZooKeeper 集群                       */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
zkServer.sh start
exit
remotessh
ssh -t hadoop@centos02 << remotessh
zkServer.sh start
exit
remotessh
ssh -t hadoop@centos03 << remotessh
zkServer.sh start
exit
remotessh
}

function start_zkfc(){
echo -e "\033[32;1m
/****************************************************************/
/*                      启动 ZKFC 守护进程                      */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
hadoop-daemon.sh start zkfc
exit
remotessh
ssh -t hadoop@centos02 << remotessh
hadoop-daemon.sh start zkfc
exit
remotessh
}

function start_hdfs(){
echo -e "\033[32;1m
/****************************************************************/
/*                        启动 HDFS HA                          */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
start-dfs.sh
exit
remotessh
}

function start_hdfs_ha(){
start_hdfs
start_zkfc
}

function start_yarn_ha(){ 
echo -e "\033[32;1m
/****************************************************************/
/*                         启动 YARN HA                         */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
start-yarn.sh
exit
remotessh
ssh -t hadoop@centos02 << remotessh
yarn-daemon.sh start resourcemanager
exit
remotessh
}

function start_hbase_ha(){
echo -e "\033[32;1m
/****************************************************************/
/*                        启动 HBase HA                         */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
start-hbase.sh
exit
remotessh
}

function start_hive(){
echo -e "\033[32;1m
/****************************************************************/
/*                          启动 Hive                           */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
hive --service metastore &
sleep 10
exit
remotessh
}

function start_spark_ha(){
echo -e "\033[32;1m
/****************************************************************/
/*                       启动 Spark HA                          */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
/opt/modules/spark-2.4.0-bin-hadoop2.7/sbin/start-all.sh
exit
remotessh
ssh -t hadoop@centos02 << remotessh
/opt/modules/spark-2.4.0-bin-hadoop2.7/sbin/start-all.sh
exit
remotessh
}

function start_kafka(){
echo -e "\033[32;1m
/****************************************************************/
/*                         启动 Kafka                           */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
/opt/modules/kafka_2.11-2.0.0/bin/kafka-server-start.sh -daemon /opt/modules/kafka_2.11-2.0.0/config/server.properties
exit
remotessh
ssh -t hadoop@centos02 << remotessh
/opt/modules/kafka_2.11-2.0.0/bin/kafka-server-start.sh -daemon /opt/modules/kafka_2.11-2.0.0/config/server.properties
exit
remotessh
ssh -t hadoop@centos03 << remotessh
/opt/modules/kafka_2.11-2.0.0/bin/kafka-server-start.sh -daemon /opt/modules/kafka_2.11-2.0.0/config/server.properties
exit
remotessh
}

function start_storm(){
echo -e "\033[32;1m
/****************************************************************/
/*                         启动 Storm                           */
/****************************************************************/\033[0m"
ssh -t hadoop@centos01 << remotessh
storm nimbus >/dev/null 2>&1 &
storm ui >/dev/null 2>&1 &
exit
remotessh
ssh -t hadoop@centos02 << remotessh
storm supervisor >/dev/null 2>&1 &
exit
remotessh
ssh -t hadoop@centos03 << remotessh
storm supervisor >/dev/null 2>&1 &
exit
remotessh
}

function start_elasticsearch(){
echo -e "\033[32;1m
/****************************************************************/
/*                     启动 Elasticsearch                       */
/****************************************************************/\033[0m"
ssh -tt hadoop@centos01 << remotessh
elasticsearch -d
sleep 30
exit
remotessh
}

function start_kibana(){
echo -e "\033[32;1m
/****************************************************************/
/*                         启动 Kibana                          */
/****************************************************************/\033[0m"
ssh -tt hadoop@centos01 << remotessh
kibana &
sleep 30
exit
remotessh
}

function start_elasticsearch_head(){
echo -e "\033[32;1m
/****************************************************************/
/*                   启动 Elasticsearch Head                    */
/****************************************************************/\033[0m"
ssh -tt hadoop@centos01 << remotessh
cd /opt/modules/elasticsearch-head-master
grunt server &
sleep 30
exit
remotessh
}

function in_hadoop(){
echo "${hadoop[@]}" | grep -wq $1 && return 0 || return 1
}

function elasticsearch_in_use(){
echo "${use[@]}" | grep -wq "elasticsearch" && return 0 || return 1
}

hadoop=(all hdfs yarn hbase hive spark kafka storm elasticsearch kibana elasticsearch_head)

use=()

if [[ $# == 0 ]]
then
help
echo -e "\033[34;1m目前你的 start_cluster.sh 后面并没有携带任何参数，请问你是否只打算启动一个 Zookeeper ? [Y/n] \033[0m \c"
read -r -p "" input
case $input in
[yY][eE][sS]|[yY])
start_zookeeper
;;
[nN][oO]|[nN])
exit 1
;;
*)
exit 1
;;
esac
else
if [[ "$1" == "-h" || "$1" == "--h" || "$1" == "-H" || "$1" == "--H" || "$1" == "-help" || "$1" == "--help" ]]
then
help
exit 1
fi
for i in $*
do
in_hadoop $i
if [ $? -eq 1 ]
then
echo -e "\033[31;1m未知参数 【 $i 】 ，请重新输入 \033[0m" 
help
exit 1
else
use+=($i)
fi
done

for i in ${use[@]}
do
if [ $i == "all" ]
then
echo -e "\033[33;1m参数中包含【all】，这将会启动所有Hadoop组件，这样做会给虚拟机以及外部本机造成很大的压力，并且有可能造成难以预料的后果；请确认你要这么做： [Y/n] \033[0m \c"
read -r -p "" confrim_all
case $confrim_all in
[yY][eE][sS]|[yY])
echo -e "\033[32;1m
/****************************************************************/
/*                         启动全部组件                         */
/****************************************************************/\033[0m"
start_zookeeper
start_hdfs_ha
start_yarn_ha
start_hbase_ha
start_hive
start_spark_ha
start_kafka
start_storm
start_elasticsearch
start_kibana
start_elasticsearch_head
exit 1
;;
[nN][oO]|[nN])
exit 1
;;
*)
exit 1
;;
esac
fi

if [ $i == "kibana" ]
then
elasticsearch_in_use
if [ $? -eq 1 ]
then
echo -e "\033[33;1mKibana的启动依赖于elasticsearch，如若想要启动kibana，请将【elasticsearch】添加到参数列表【kibana】之前或去掉【kibana】\033[0m"
exit 1
fi
fi
if [ $i == "elasticsearch_head" ]
then
elasticsearch_in_use
if [ $? -eq 1 ]
then
echo -e "\033[33;1mElasticsearch Head的启动依赖于elasticsearch，如若想要启动kibana，请将【elasticsearch】添加到参数列表【elasticsearch_head】之前或去掉【elasticsearch_head】\033[0m"
exit 1
fi
fi
done
fi

start_zookeeper
for i in ${use[@]}
do
if [ $i == "hdfs" ]
then
start_hdfs_ha
fi
if [ $i == "yarn" ]
then
start_yarn_ha
fi
if [ $i == "hbase" ]
then
start_hbase_ha
fi
if [ $i == "hive" ]
then
start_hive
fi
if [ $i == "spark" ]
then
start_spark_ha
fi
if [ $i == "kafka" ]
then
start_kafka
fi
if [ $i == "storm" ]
then
start_storm
fi
if [ $i == "elasticsearch" ]
then
start_elasticsearch
fi
if [ $i == "kibana" ]
then
start_kibana
fi
if [ $i == "elasticsearch_head" ]
then
start_elasticsearch_head
fi
done

集群关闭脚本stop-cluster.sh：

ssh hadoop@centos01 << remotessh
stop-hbase.sh
/opt/modules/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
/opt/modules/spark-2.4.0-bin-hadoop2.7/sbin/stop-all.sh
stop-yarn.sh
hadoop-daemon.sh stop zkfc
stop-dfs.sh
zkServer.sh stop
exit
remotessh

ssh hadoop@centos02 << remotessh
/opt/modules/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
yarn-daemon.sh stop resourcemanager
/opt/modules/spark-2.4.0-bin-hadoop2.7/sbin/stop-all.sh
hadoop-daemon.sh stop zkfc
zkServer.sh stop
exit
remotessh

ssh hadoop@centos03 << remotessh
/opt/modules/kafka_2.11-2.0.0/bin/kafka-server-stop.sh
zkServer.sh stop
exit
remotessh

ssh hadoop@centos01 << remotessh
pkill -f HMaster
pkill -f HRegionServer
pkill -f Kafka
pkill -f Worker
pkill -f Master
pkill -f NodeManager
pkill -f ResourceManager
pkill -f RunJar
pkill -f DFSZKFailoverController
pkill -f DataNode
pkill -f NameNode
pkill -f JournalNode
pkill -f QuorumPeerMain
pkill -f nimbus
pkill -f core
pkill -f node
pkill -f Elasticsearch
pkill -f grunt
exit
remotessh

ssh hadoop@centos02 << remotessh
pkill -f HMaster
pkill -f HRegionServer
pkill -f Kafka
pkill -f Worker
pkill -f Master
pkill -f NodeManager
pkill -f ResourceManager
pkill -f RunJar
pkill -f DFSZKFailoverController
pkill -f DataNode
pkill -f NameNode
pkill -f JournalNode
pkill -f QuorumPeerMain
exit
remotessh

ssh hadoop@centos03 << remotessh
pkill -f HMaster
pkill -f HRegionServer
pkill -f Kafka
pkill -f Worker
pkill -f Master
pkill -f NodeManager
pkill -f ResourceManager
pkill -f RunJar
pkill -f DFSZKFailoverController
pkill -f DataNode
pkill -f NameNode
pkill -f JournalNode
pkill -f QuorumPeerMain
exit
remotessh

通过-h参数来获取集群启动信息：

Hadoop组件可选，默认只启动一个ZooKeeper

上面的UI中的”一键启动集群”由于内存限制，并没有启动Elasticsearch及其附带组件，启动的是hdfs、yarn、hbase、hive、spark、kafka、storm；并且其实UI界面也并没有用到这么多功能，只用到了hdfs、hbase、kafka、spark而已，即执行了：

./start-cluster.sh hdfs kafka hbase yarn spark storm

同理，“一键停止集群也是相当于执行了：

./stop-cluster.sh

2.4、本机MySQL配置

在Kafka界面中的SparkStreamingWordCount程序，实际上是经过scala处理后将结果写入本机MySQL数据库：

因此需要新建一个数据库用来存放WordCount结果，并且最好不要用root用户（这里我用的之前新建的名为hive的用户）

mysql -u root -p 

create user hive@'localhost' identified by 'hive';
update mysql.user set host = '%' where user = 'hive';
FLUSH PRIVILEGES;
GRANT ALL PRIVILEGES ON *.* TO 'hive'@'%' WITH GRANT OPTION;

mysql -u hive -p # 密码也是hive

create database spark;
use spark;
create table wordcount(word varchar(100),count int);

创建了一个名为“spark”的数据库并在该数据库中建造了一张名为“wordcount”的表

2.5、Scala - StreamingKafkaWordCount 自动开关

首先在IDEA中启动StreamingKafkaWordCount程序，然后将其中的内容展开复制下来替换掉config包下的InitConfig.java中的private static final String wordCountCmd中的内容：

2.6、配置本机 jps 命令

首先获取到本机java安装路径下的 jps.exe 执行文件的路径：

然后将其中的内容展开复制下来替换掉config包下的InitConfig.java中的private static final String jpsCmd中的内容：

2.7、Kafka集群创建主题

创建主题topictest和topictest2

/opt/modules/kafka_2.11-2.0.0/bin/kafka-topics.sh --create --zookeeper centos01:2181,centos02:2181,centos03:2181 --replication-factor 2 --partitions 2 --topic topictest

/opt/modules/kafka_2.11-2.0.0/bin/kafka-topics.sh --create --zookeeper centos01:2181,centos02:2181,centos03:2181 --replication-factor 2 --partitions 2 --topic topictest2

3、源码概览

3.1、源码结构

3.2、配置文件包 config

3.2.1、InitConfig

通过配置静态变量wordCountCmd、jpsCmd并将其getter方法作为Bean进行注入，需要使用时将其自动装配即可：

@Autowired
private InitConfig initConfig;

/* ... */
initConfig.getWordCountCmd()
initConfig.getJpsCmd()
/* ... */

这里不能通过application.yml配置文件进行注入，因为项目在启动时就需要启动StreamingKafkaWordCount，而此时还不能通过@Value注解来获取配置文件中的值，因此只能通过配置类中的静态变量来获取

3.3、controller 层

• 大部分页面都采用了前后端分离，只有HBase的详情页面没有采用前后端分离，因此除了页面控制器PageController采用@Controller注解，其余所有控制器都采用的@RestController注解且请求URL都以“/api/”开头

• 所有有@RestController的类都有一个静态内部类继承了自定义的utils工具包中的ResultMapToJson来返回统一格式的json数据作为请求的返回值

• 静态起源的请求都通用的@GetMapping，其他请求同意用的@PostMapping，没有使用@PutMapping、@DeleteMapping等（因为前端请求时可以统一使用$.post解决，就不需要麻烦的使用$.ajax进行请求还要指定请求类型）

3.3.1、PageController

所有方法都加了@GetMapping注解用来接受页面请求，返回resource资源文件夹中的templates文件夹中的html页面

3.3.2、ShellController

• 处理/shell页面发来的请求

• 具体实现通过自定义的utils工具包中的ShellUtil来实现

3.3.3、HDFSController

• 处理/hdfs页面发来的请求

• 实现了一些基本的HDFS操作，根据不同的请求来执行不同的任务

3.3.4、KafkaController

• 处理/kafka页面发来的请求

• 实现了基本的Kafka操作以及调用scala编写的`StreamingKafkaWordCount程序

• StreamingKafkaWordCount程序的启动和停止通过自定义的utils工具包中的ScalaUtil来实现（再进一步是通过Runtime.getRuntime().exec()执行命令来实现的）

• 清理数据库通过自定义的utils工具包中的MysqlUtil来实现

3.3.5、MapReduceController

• 处理/mapreduce页面发来的请求执行单词计数
• 具体实现通过自定义的utils工具包中的WordCount来实现

3.3.6、HBaseController

• 处理/hbase页面发来的请求
• 实现了一些基本的HBase操作，根据不同的请求来执行不同的任务
• 其中HBase连接的Connection实例的基础配置通过自定义的utils工具包中的HBaseUtil来实现

3.4、utils 工具包

utils工具包中的所有的类都添加了@Component注解注入到spring容器中

3.4.1、ShellConnectionSignal

• 作为java连接集群操作Shell脚本的单例类，供ShellUtil使用

• 通过com.jcraft.jsch包实现java连接远程Linux主机并操作shell脚本

• 私有属性session、exec及其getter、setter方法供ShellUtil创建连接实例

3.4.2、ShellUtil

• java连接linux服务器并操作shell脚本的工具类
• execCommand(String host, String command)方法用来在host主机上执行command命令，执行结果返回一个String类型的List，调用者可将其转为json格式用作请求的返回值
• 通过调用ShellConnectionSignal的session、exec的setter、getter方法创建实例后调用该实例的connect()方法创建连接，执行完毕后调用close()方法关闭连接

3.4.3、MySQLConnectionSignal

• 作为java连接本机mysql数据库的单例类，供MySqlUtil使用
• 私有属性connection及其getter、setter方法供MysqlUtil创建连接实例

3.4.4、MysqlUtil

• 私有final变量mysql为MySQLConnectionSignal的实例

• 其getConnection()方法通过调用mysql的setConnection()方法设置mysql连接对象所需属性后，返回一个mysql的getConnection()方法的返回对象，进而 获得一个java.sql.Connection连接对象去执行操作

3.4.5、WordCount

• MapReduce实现单词计数，返回String类型的结果，形如[(xxx, 1),(yyy, 2), ... , (String, int)]

3.4.6、HBaseUtil

• java操作HBase数据库的工具类
• 通过getConnection()获得一个org.apache.hadoop.hbase.client.Connection对象的实例，然后再进行相应的操作

3.4.7、ListUtil

• 目前只有一个静态方法getCount(List<String> list, String s)，用来判断字符串s在列表list中出现的次数，返回值类型为int

3.4.8、ResultMapToJson

• 抽象类

• 对com.alibaba.fastjson.JSON的toJSONString()方法做了进一步的封装，getResult()方法返回形如{"msg":"success/error", "detail":"..."}的统一格式数据，供前端请求时进行响应；后端代码都包含在try语句中，若出现异常，则会将e.toString()赋值给detail，前端则会显示出报错而不会500，例如：

3.4.9、ScalaUtil

• 通过java程序调用scala程序`StreamingKafkaWordCount的桥梁

• 内涵两个方法startStreamingKafkaWordCount()和stopStreamingKafkaWordCount()分别是启动和关闭`StreamingKafkaWordCount程序

• 启动是通过执行InitConfig的wordCountCmd命令；终止则是先通过InitConfig的jpsCmd获取到StreamingKafkaWordCount的端口然后通过taskkill命令将其杀死；他们的执行都是通过Runtime.getRuntime().exec()实现的

• /kafka页面中的【清空】按钮就是调用了stopStreamingKafkaWordCount()方法先停止StreamingKafkaWordCount程序，清空mysql数据库后再调用startStreamingKafkaWordCount()程序启动StreamingKafkaWordCount()程序

3.5、启动类 BigDataApplication

• @SpringBootApplication注解标志着这个类是程序的启动类
• System.setProperty("HADOOP_USER_NAME", "hadoop");指明了该程序对Hadoop集群的一切操作的权限都是hadoop用户而非root
• Runtime.getRuntime().exec(initConfig.getWordCountCmd());表示程序在启动之初就会执行StreamingKafkaWordCount程序

3.6、SparkStreaming + Kafka 单词计数类 StreamingKafkaWordCount

• StreamingKafkaWordCount是由scala编写的一个类
• 将单词计数结果写入Mysql数据库，每秒一次
• 被ScalaUtil调用执行

4、源码下载

• 下载地址一：http://www.philpy.top/download/

• 下载地址二：https://github.com/sleepyyoung/Hadoop-Clutser-Manager/

原文链接:https://blog.csdn.net/qq_43613793/article/details/117303208