集群管理相关脚本汇总

偏执的太偏执、 2022-12-27 09:25 117阅读 0赞

### 目录 ###

*   *  00 查看集群各主机进程（xcall.sh）
     *  01 数据分发同步脚本xsync
     *  02 集群启动停止管理脚本mycluster
     *  03 Zookeeper启动停止管理脚本
     *  04 Kafak启动停止管理脚本
     *  05 Flume采集启动停止脚本
     *  06 数据生成脚本
     *  07 Hive服务启动停止管理脚本
     *  08 Sqoop采集MySQL数据到HDFS脚本
     *  09 清除相关数据

## 00 查看集群各主机进程（xcall.sh） ##

#!/bin/bash
    for host in hadoop102 hadoop103 hadoop104
    do
            echo "=======================$host================"
            ssh $host "$*"
    done

> 使用示例： xcall.sh jps

## 01 数据分发同步脚本xsync ##

#1、判断是否传入需要同步的目录/文件
    if [ $# -lt 1 ]
    then
    	echo "必须传入一个同步的文件或者目录!!"
    	exit
    fi
    #2、遍历需要同步的多个目录/文件
    #$*: 将输入的所有当做一个整体
    # xsync java hadoop
    #$@: 将输入参数当做单独的个体
    for f in $@
    do
    	#3、判断输入的目录/文件是否存在
    	#-f: 判断是否为文件
    	#-e: 判断文件/目录是否存在
    	#-d: 判断是否为目录
    	if [ -e $f ]
    	then
    		#4、获取父目录以及当前需要同步的目录/文件名
    		pdir=$(cd $(dirname $f);pwd)
    		cname=$(basename $f)
    		#5、遍历主机创建父目录
    		for host in hadoop102 hadoop103 hadoop104
    		do
    			ssh $host "mkdir -p $pdir"
    			#6、同步目录/文件
    			rsync -av $pdir/$cname $host:$pdir
    		done
    	fi
    done

## 02 集群启动停止管理脚本mycluster ##

#! /bin/bash
    #1、判断是否输入参数
    if [ $# -lt 1 ]
    then
    	echo "必须传入一个参数..."
    	exit
    fi
    #2、根据参数匹配执行对应的逻辑
    case $1 in
    "start")
    	echo "======================start hdfs====================="
    	ssh hadoop102 "start-dfs.sh"
    	echo "======================start yarn====================="
    	ssh hadoop103 "start-yarn.sh"
    ;;
    "stop")
    	echo "=====================stop hdfs======================="
    	ssh hadoop102 "stop-dfs.sh"
    	echo "=====================stop yarn======================="
    	ssh hadoop103 "stop-yarn.sh"
    ;;
    *)
    	echo "参数输入错误....."
    ;;
    esac

## 03 Zookeeper启动停止管理脚本 ##

#! /bin/bash
    #1、判断参数是否输入
    if [ $# -lt 1 ]
    then
    	echo "请输入参数...."
    	exit
    fi
    #2、根据参数执行逻辑
    case $1 in
    "start")
    	for host in hadoop102 hadoop103 hadoop104
    	do
    		echo "==================start $host zookeeper==============="
    		ssh $host "/opt/module/zookeeper/bin/zkServer.sh start"
    	done
    ;;
    "stop")
    	for host in hadoop102 hadoop103 hadoop104
    	do
    		echo "==================stop $host zookeeper==============="
    		ssh $host "/opt/module/zookeeper/bin/zkServer.sh stop"
    	done
    ;;
    "status")
    	for host in hadoop102 hadoop103 hadoop104
    	do
    		echo "==================status $host zookeeper==============="
    		ssh $host "/opt/module/zookeeper/bin/zkServer.sh status"
    	done
    ;;
    *)
    	echo "参数输入错误...."
    ;;
    esac

## 04 Kafak启动停止管理脚本 ##

#1、判断是否传入参数
    if [ $# -lt 1 ]
    then
    	echo "必须传入一个参数..."
    	exit
    fi
    #2、根据传入的参数执行逻辑
    case $1 in
    "start")
    	for host in hadoop102 hadoop103 hadoop104
    	do
    		echo "====================$host================="
    		ssh $host "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"
    	done
    ;;
    "stop")
    	for host in hadoop102 hadoop103 hadoop104
    	do
    		echo "====================$host================="
    		ssh $host "/opt/module/kafka/bin/kafka-server-stop.sh"
    	done
    ;;
    "statu")
    	for host in hadoop102 hadoop103 hadoop104
    	do
    		echo "====================$host================="
    		pid=$(ssh $host "ps -ef |grep server.properties|grep -v grep")
    		[ "$pid" ] && echo "kafka进程正常" || echo "kafka进程不存在或者异常"
    	done
    ;;
    *)
    	echo "参数传入错误...."
    ;;
    esac

## 05 Flume采集启动停止脚本 ##

#! /bin/bash
    #1、判断参数是否传入
    if [ $# -lt 1 ]
    then
    	echo "必须传入参数....."
    	exit
    fi
    #2、根据传入的参数匹配逻辑
    case $1 in
    "start")
    	for host in hadoop102 hadoop103
    	do
    		echo "==============start $host log to kafka==========="
    		ssh $host "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf/ -f /opt/module/flume/job/log-to-kafka.conf >/opt/module/flume/logs/app.log 2>&1 &"
    	done
    ;;
    "stop")
    	for host in hadoop102 hadoop103
    	do
    		echo "=============stop $host log to kafka=============="
    		ssh $host "ps -ef |grep log-to-kafka.conf |grep -v grep|awk '{print \$2}'|xargs kill -9"
    	done
    ;;
    *)
    	echo "参数传入错误...."
    ;;
    esac

## 06 数据生成脚本 ##

#! /bin/bash
    for host in hadoop102 hadoop103
    do
    	echo "====================start $host data===================="
    	#标准输入: 0<
    	# position.sh < a.txt
    	#标准输出: 1
    	# position.sh > /xx.txt
    	#错误输出: 2
    	# position.sh >/xx.txt 2> &1
    	# nohup: 免挂断,ssh连接断开不会停止程序
    	# &: 后台执行,ssh连接断开程序会终止
    	ssh $host "nohup java -jar /opt/module/applog/gmall2020-mock-log-2020-04-01.jar >/dev/null 2>&1 &"
    done

## 07 Hive服务启动停止管理脚本 ##

#!/bin/bash
    HIVE_LOG_DIR=$HIVE_HOME/logs
    if [ ! -d $HIVE_LOG_DIR ]
    then
            mkdir -p $HIVE_LOG_DIR
    fi
    #检查进程是否运行正常，参数1为进程名，参数2为进程端口
    function check_process()
    { 
        pid=$(ps -ef 2>/dev/null | grep -v grep | grep -i $1 | awk '{print $2}')
        ppid=$(netstat -nltp 2>/dev/null | grep $2 | awk '{print $7}' | cut -d '/' -f 1)
        echo $pid
        [[ "$pid" =~ "$ppid" ]] && [ "$ppid" ] && return 0 || return 1
    }
    
    function hive_start()
    { 
        metapid=$(check_process HiveMetastore 9083)
        cmd="nohup hive --service metastore >$HIVE_LOG_DIR/metastore.log 2>&1 &"
        cmd=$cmd" sleep 4; hdfs dfsadmin -safemode wait >/dev/null 2>&1"
        [ -z "$metapid" ] && eval $cmd || echo "Metastroe服务已启动"
        server2pid=$(check_process HiveServer2 10000)
        cmd="nohup hive --service hiveserver2 >$HIVE_LOG_DIR/hiveServer2.log 2>&1 &"
        [ -z "$server2pid" ] && eval $cmd || echo "HiveServer2服务已启动"
    }
    
    function hive_stop()
    { 
        metapid=$(check_process HiveMetastore 9083)
        [ "$metapid" ] && kill $metapid || echo "Metastore服务未启动"
        server2pid=$(check_process HiveServer2 10000)
        [ "$server2pid" ] && kill $server2pid || echo "HiveServer2服务未启动"
    }
    
    
    case $1 in
    "start")
        hive_start
        ;;
    "stop")
        hive_stop
        ;;
    "restart")
        hive_stop
        sleep 2
        hive_start
        ;;
    "status")
        check_process HiveMetastore 9083 >/dev/null && echo "Metastore服务运行正常" || echo "Metastore服务运行异常"
        check_process HiveServer2 10000 >/dev/null && echo "HiveServer2服务运行正常" || echo "HiveServer2服务运行异常"
        ;;
    *)
        echo Invalid Args!
        echo 'Usage: '$(basename $0)' start|stop|restart|status'
        ;;
    esac

## 08 Sqoop采集MySQL数据到HDFS脚本 ##

#1.判断传入参数
    [ "$2" ] && datastr=$2 || datastr= $(date -d '-1 day' +%Y%m%d)
    
    import_data(){ 
    /opt/module/sqoop/bin/sqoop import \
    --connect jdbc:mysql://hadoop102:3306/gmall  \
    --username root \
    --password  123321 \
    --delete-target-dir  \
    --num-mappers 1  \
    --query  "$1"  \
    --split-by  $3  \
    --target-dir hdfs://hadoop102:8020/gmall/$2/$datastr  \
    --compress  \
    --compression-codec  lzop  \
    --null-string  '\\N'  \
    --null-non-string  '\\N'  \
    --fields-terminated-by  ,
    }
    
    hadoop jar /opt/module/hadoop/share/hadoop/common/hadoop-lzo-0.4.20.jar com.hadoop.compression.lzo.DistributedLzoIndexer hdfs://hadoop102:8020/gmall/$2/$datastr
    
    
    import_sku_info(){ 
    	import_data "select * from sku_info where \$CONDITIONS "  sku_info id
    }
    
    import_base_dic(){ 
    	import_data "select * from base_dic where \$CONDITIONS "  base_dic dic_code
    }
    
    
    import_base_province(){ 
    	import_data "select * from base_province where \$CONDITIONS "  base_province id
    }
    
    
    import_base_region(){ 
    	import_data "select * from base_region where \$CONDITIONS "  base_region id
    }
    
    
    import_order_detail_first(){ 
    	import_data "select * from order_detail where \$CONDITIONS "  order_detail id
    }
    
    import_order_detail_all(){ 
    	import_data "select * from order_detail where DATE_FORMAT(create_time,'%Y%m%d') = '$datastr' and \$CONDITIONS "  order_detail id
    }
    
    
    import_order_info_first(){ 
    	import_data "select * from order_info where \$CONDITIONS "  order_info id
    }
    
    import_order_info_all(){ 
    	import_data "select * from order_info where DATE_FORMAT(create_time,'%Y%m%d') = '$datastr' or DATE_FORMAT(operate_time,'%Y%m%d') and \$CONDITIONS "  order_info id
    }
    
    
    
    
    #2.根据传入参数导入数据
    case $1 in
    "first")
    import_sku_info
    import_base_dic
    import_base_province
    import_base_region
    import_order_detail_first
    import_order_info_first
    ;;
    "all")
    import_sku_info
    import_base_dic
    import_order_detail_all
    import_order_info_all
    ;;
    "sku_info")
    import_sku_info
    ;;
    "base_dic")
    import_base_dic
    ;;
    "base_province")
    import_base_province
    ;;
    "base_region")
    import_base_region
    ;;
    "order_detail")
    import_order_detail_all
    ;;
    "order_info")
    import_order_info_all
    ;;
    esac

## 09 清除相关数据 ##

#!/bin/bash
    for host in hadoop102 hadoop103 hadoop104
    do
            ssh $host rm -rf $HADOOP_HOME/data $HADOOP_HOME/logs
            ssh $host sudo rm -rf /tmp/*
    done