Spark集群搭建

  
Spark集群搭建

一、基础集群脚本

1、长途调用脚本(remote_call.sh)

如果有传命令参数，则执行该命令；如果没有传命令参数，则不执行。

1. #!/bin/bash
3. cmd=$1
4. if [ ! $cmd ];then
5.         cmd="jps"
6. fi
8. # 提取集群免密通信的虚拟机主机名
9. hosts=`sed -n '3,$p' /etc/hosts | awk '{print $2}'`
11. # 遍历所有主机
12. for host in $hosts;do
13.         echo "-------- $host --------"
14.         # 此处使用"$cmd"的原因是避免将命令中的空格识别为多条命令
15.         ssh root@$host "$cmd"
16. done

复制代码

2、长途复制目次脚本(remote_copy.sh)

起首，要验证待复制的目次在本机是否存在；然后需要从/etc/hosts文件中获取撤消当前主机名的其他主机名，并且对每个主机举行循环操作，先判断父目次是否存在，再举行递归复制。

1. #!/bin/bash
3. path=$1 # 本机目录
5. # 验证路径是否存在
6. if [ ! -e $path ];then
7.         echo "目录 $path 不存在，无法拷贝"
8.         exit 0
9. fi
11. # 提取集群中排除当前主机名之外的所有主机名
12. # 只需验证父目录是否存在即可
13. me=`hostname`
14. parent=`dirname $path`
15. hosts=`sed -n '3,$p' /etc/hosts | awk '{print $2}' | grep -v "$me"`
17. for host in $hosts;do
18.         scp -r $path root@$host:$parent
19.         echo "-------- 拷贝 $path 到 $host 成功 --------"
20. done

复制代码

3、Spark集群服务管理脚本

按照"先开的后关"的原则，按照"ZooKeeper-HDFS-YARN-Spark-启用其他Master节点"的次序举行集群的服务管理

1. #!/bin/bash
3. if [ "$1" == "start" ];then
4.         # 自定义启动服务函数，传入两个字符串参数：参数1为启动命令，参数2为启动方式
5.     function startService(){
6.                 # 定义两个局部遍历依次接收用户传入的两个参数
7.                 # service 为启动命令
8.                 local service=$1
9.                 # exeType 为启动方式(daemon:后台启动(用于那些会占用窗口的服务)，非daemon表示正常启动)
10.                 local exeType=$2
11.                 if [ "$exeType" == "daemon" ];then
12.                         # 后台启动
13.                         nohup $service 1>/dev/null 2>&1 &
14.                 else
15.                         # 正常启动
16.                         $service 1>/dev/null 2>&1
17.                 fi
18.                 # 判断执行启动脚本后的状态
19.                 if [ $? -ne 0 ];then
20.                         # 因为服务按从上到下有依赖关系，所以任何一个脚本执行出错，将退出整个脚本，所以选择exit而非return
21.                         # 函数按通用规则 return 0:正常，非0:表示异常；若函数中return缺省，则默认返回最后一条命名的状态即改命令执行后的$?
22.                         echo "ERROR : FAIL TO EXECUTE $service IN MODE OF $exeType ... EXIT ..."
23.                         # 退出脚本
24.                         exit 1
25.                 else
26.                         # 成功则正常输出提示结果
27.                         echo "INFO : SUCCESS TO EXECUTE $service IN MODE OF $exeType"
28.                 fi
29.     }
31.         # 依次按服务依赖顺序调用启动服务函数，并为每次调用传入启动命令和启动方式
32.         # 启动 Zookeeper
33.         /root/install/remote_call.sh "zkServer.sh start"
34.         # 启动 hadoop hdfs
35.     startService "start-dfs.sh" "non-daemon"
36.         # 启动 hadoop yarn
37.     startService "start-yarn.sh" "non-daemon"
38.         # 启动 Spark
39.         startService "start-all.sh" "non-daemon"
40.         # 实现集群高可用
41.         ssh root@master02 "start-master.sh"
43. elif [ "$1" == "stop" ];then
44.         # 自定义启动服务函数，传入两个字符串参数：参数1为关闭命令，参数2为关闭方式
45.     function stopService(){
46.                 # 服务名称或关闭命令
47.                 local service=$1
48.                 # 关闭方式(command:命令关闭，pid:根据服务查找进程编号(pid)在借助kill命令关闭)
49.                 local exeType=$2
50.                 if [ "$exeType" == "command" ];then
51.                         # 直接执行参数一命令关闭服务
52.                         $service 1>/dev/null 2>&1
53.                         # 根据关闭命令执行的状态展示结果
54.                         if [ $? -eq  0 ];then
55.                                         echo "INFO : SUCCESS TO EXECUTE $service"
56.                         else
57.                                         echo "ERROR : FAIL TO EXECUTE $service"
58.                         fi
59.                 else
60.                         # 根据参数一传入的服务名称查看进程编号(pid)
61.                         local pid=$(jps -ml|grep $service|awk '{print $1}')
62.                         if [ "$pid" ];then
63.                                 # 如果进程编号存在，则直接强制 kill
64.                                 kill -9 $pid
65.                                 # 根据kill的状态展示结果
66.                                 if [ $? -eq  0 ];then
67.                                                         echo "INFO : SUCCESS TO KILL $service WITH PID $pid"
68.                                         else
69.                                                         echo "ERROR : FAIL TO KILL $service WITH PID $pid"
70.                                         fi
71.                         else
72.                                 echo "INFO : NO SERVICE EXIST WITH NAME OF $service"
73.                         fi
74.                 fi
75.     }
77.         # 根据服务的依赖关系，逆向逐个关闭服务
78.         # 关闭高可用另外起的master
79.         ssh root@master02 "stop-master.sh"
80.         # 关闭Spark
81.         stopService "stop-all.sh" "command"
82.         # 最后关闭 yarn 和 hdfs
83.     stopService "stop-yarn.sh" "command"
84.     stopService "stop-dfs.sh" "command"
85.         # 关闭ZooKeeper
86.         /root/install/remote_call.sh "zkServer.sh stop"
87. else
88.         # 附带查一下java进程
89.     jps -ml
90. fi

复制代码

如果需要切换主节点的状态，可以先通过stop-master.sh使A-Master节点状态由Active->Standby，待B-Master根据机制切换为Active态之后再重启A-Master节点。
二、配置相关文件

1. cd /opt/software/spark-3.1.2/conf
2. // mv xxx.template xxx 将所有template文件去除后缀为正式配置文件
3. vim spark-env.sh
4. -------------------------------------------------------------------
5. export JAVA_HOME=/opt/software/jdk1.8.0_171
7. export HADOOP_CONF_DIR=/opt/software/hadoop-3.1.3/etc/hadoop
9. SPARK_MASTER_WEBUI_PORT=9090
11. SPARK_HISTORY_OPTS="
12. -Dspark.history.ui.port=9091
13. -Dspark.history.fs.logDirectory=hdfs://master01:8020/spark_event_log_dir
14. -Dspark.history.retainedApplications=30
15. "
17. SPARK_DAEMON_JAVA_OPTS="
18. -Dspark.deploy.recoveryMode=ZOOKEEPER
19. -Dspark.deploy.zookeeper.url=master01,master02,worker01
20. -Dspark.deploy.zookeeper.dir=/spark
21. "
22. -------------------------------------------------------------------
24. vim workers // 配置集群所有机器名称
25. -------------------------------------------------------------------
26. master01
27. master02
28. worker01
29. -------------------------------------------------------------------
31. vim spark-default.conf
32. -------------------------------------------------------------------
33. spark.eventLog.enabled           true
34. spark.eventLog.dir               hdfs://master01:8020/spark_event_log_dir
35. spark.yarn.historyServer.address=master01:9091
36. spark.history.ui.port=9091
37. -------------------------------------------------------------------
39. vim yarn-site.xml // [选配]：在启动Yarn模式时才需要配置
40. -------------------------------------------------------------------
41. <property>
42.     <name>yarn.nodemanager.pmem-check-enabled</name>
43.     <value>false</value>
44. </property>
45. <property>
46.     <name>yarn.nodemanager.vmem-check-enabled</name>
47.     <value>false</value>
48. </property>
49. -------------------------------------------------------------------
51. ./remote_copy.sh /opt/software/spark-3.1.2/ // 拷贝Spark内容
52. ./remote_copy.sh /etc/profile.d/myenv.sh // 拷贝环境变量
53. ./remote_call.sh "source /etc/profile"
54. ./remote_call.sh "/opt/software/service-bigdata.sh start" // 完整启动Spark集群中的所有服务

复制代码

在此配置下，有以下关键信息：

• 端口配置
  
  
  
  • Spark Master Web UI：9090
    
  • Spark HistoryServer UI：9091
    
  
  
• 日志文件路径
  
  
  
  • 日志文件路径：hdfs://master01:8020/spark_event_log_dir
    
  
  
• 关闭内存检查的原因：Spark使命吃资源，不能为了保证资源不被太过利用就强行停止Spark使命。
  

三、使命提交案例

1. 案例详解

ProductsAnalyzer类
val path = args(0) val dirPath = args(1) 表示输入和输出路径在提交使命时具体指定。

1. package envtest
3. import core.{SparkSessionBuilder, SparkSessionBuilderDeployer}
4. import org.apache.spark.SparkContext
5. import org.apache.spark.rdd.RDD
6. import org.apache.spark.sql.expressions.Window
8. object ProductsAnalyzer {
9.   case class Product(name: String, price: Double, date: String, market: String, province: String, city: String)
10.   implicit class StrExt(line: String){
11.     // 香菜        2.80        2018/1/1        山西汾阳市晋阳农副产品批发市场        山西        汾阳
12.     val regexProduct = "(.*?)\t(.*?)\t(.*?)\t(.*?)\t(.*?)\t(.*?)".r
13.     def toProduct: Product = {
14.       line match {
15.         // 使用模式匹配将字符串和正则表达式进行匹配，如果成功，则将各个匹配到的部分转换为Product对象并进行返回
16.         // java.lang.NumberFormatException: empty String:输入数据中的某些行可能不符合预期的格式，导致 toDouble 转换失败。解决方案是：使其转化为Product对象时，如果为空，则转化为0.0f
17.         case regexProduct(name, price, date, market, province, city) => Product(name, if (price.trim.isEmpty) 0.0f else price.toDouble, date, market, province, city)
18.         case _ => throw new RuntimeException(s"产品数据格式有误：$line")
19.       }
20.     }
21.   }
22.   def main(args: Array[String]): Unit = {
23.     /**
24.      * 在Edit Configurations中指定参数，通过args[0]和args[1]获取输入路径和输出路径
25.      */
26.     val path = args(0)
27.     val dirPath = args(1)
29.     val ssb: SparkSessionBuilderDeployer = SparkSessionBuilderDeployer()
30.     val spark = ssb.spark
31.     val sc: SparkContext = ssb.sc
33.     val rddProduct: RDD[Product] = sc.textFile(path)
34.       .mapPartitions(_.map(_.toProduct))
36.     import org.apache.spark.sql.functions._
37.     import spark.implicits._
38.     // 统计每个省份下每个商品的平均价格和商品数量，并且在不同省份下按照商品数量创建排名指标
39.     spark.createDataFrame(rddProduct)
40.       .groupBy($"province",$"name")
41.       .agg(
42.         avg($"price").as("avg_price"),
43.         count("*").as("product_cnt")
44.       )
45.       .select($"province",$"name",$"avg_price",$"product_cnt",dense_rank().over(Window.partitionBy($"province").orderBy($"avg_price".desc)).as("rank"))
46.       .repartition(1)
47.       .write
48.       .mode(saveMode = "overwrite")
49.       .option("separator",",")
50.       .option("header", "true")
51.       .csv(dirPath)
53.     spark.stop()
54.   }
55. }

复制代码

SparkSessionBuilderDeployer类
SparkSessionBuilderDeployer() 使使命提交案例关于运行模式和名称的配置在使命提交时再举行设置，提升案例的重用性。

1. package core
3. import org.apache.spark.sql.SparkSession
4. import org.apache.spark.{SparkConf, SparkContext}
6. // 封装SparkSession的创建方法
7. // SparkSessionBuilderDeployer() 使任务提交案例关于运行模式和名称的配置在任务提交时再进行设置，提升案例的重用性。
8. class SparkSessionBuilderDeployer(){
9.   lazy val config:SparkConf = {
10.     new SparkConf()
11.   }
13.   lazy val spark:SparkSession = {
14.     SparkSession.builder()
15.       .config(config)
16.       .getOrCreate()
17.   }
19.   lazy val sc:SparkContext = {
20.     spark.sparkContext
21.   }
23.   def stop(): Unit = {
24.     if (null != spark) {
25.       spark.stop()
26.     }
27.   }
28. }
29. object SparkSessionBuilderDeployer {
30.   def apply(): SparkSessionBuilderDeployer = new SparkSessionBuilderDeployer
31. }

复制代码

在具体测试案例代码时，可以先通过对path和dirpath指定具体路径的方式来测试整体逻辑，再通过Edit Configurations设置具体参数来测试传参后的程序逻辑，测试完毕之后，即可直接将输入和输出路径分别修改为args(0)和args(1)，并且直接打包发布到假造机，供给Spark集群举行执行。
2. Spark Submit 选项详解

• 主要选项
  
  
  
  • –class :
    
    
    
    • 指定应用程序的主类，即包含 main 方法的类。
      
    • 示例：--class envtest.ProductsAnalyzer
      
    
    
  • –master :
    
    
    
    • 指定 Spark Master 节点的 URL。
      
    • 示例：
      
      
      
      • Standalone 模式：
        
        1. --master spark://master01:7077

        复制代码
        
        
        • 表示 Spark Master 节点运行在 master01 主机的 7077 端口上。
          
        
        
      • YARN 模式：
        
        1. --master yarn

        复制代码
        
        
        • 利用 YARN 作为资源管理器。
          
        
        
      
      
    
    
  • –deploy-mode :
    
    
    
    • 指定应用程序的运行模式。
      
    • 模式：
      
      
      
      • client：Driver 程序运行在提交使命的当地呆板上。
        
      • cluster：Driver 程序运行在集群的某个节点上。
        
      
      
    • 示例：--deploy-mode cluster
      
    
    
  • –name :
    
    
    
    • 指定应用程序的名称。
      
    • 示例：--name spark-sql-product
      
    
    
  • :
    
    
    
    • Spark 应用程序的 JAR 文件的路径，包含了主类及其依靠。
      
    • 示例：/root/spark/spark_sql-1.0.jar
      
    
    
  • <application-argument(s)>:
    
    
    
    • 应用程序的输入参数。
      
    • 示例：
      
      
      
      • 输入路径：hdfs://master02:8020/spark/data/products.txt
        
      • 输出路径：hdfs://master02:8020/spark/result/product_result
        
      
      
    
    
  
  
• 具体代码
  

standalone

1. spark-submit \--class envtest.ProductsAnalyzer \--master spark://master01:7077
2. \--deploy-mode client \--name spark-sql-product \/root/spark/spark_sql-1.0.jar \hdfs://master02:8020/spark/data/products.txt \hdfs://master02:8020/spark/result/product_result

复制代码

yarn-cluster

1. spark-submit \--class envtest.ProductsAnalyzer \--master yarn
2. \--deploy-mode cluster \--name spark-sql-product \/root/spark/spark_sql-1.0.jar \hdfs://master02:8020/spark/data/products.txt \hdfs://master02:8020/spark/result/product_result

复制代码

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