Spark（三）角色和搭建
一、Spark集群角色介绍
详见JerryLead/SparkInternals，他的图解介绍能清晰的讲清楚Spark集群
二、集群的搭建
2.1.架构(图片来源，Spark官网)
一个Driver Program含有一个SparkContext，课由ClusterManager进行通讯到Worker节点，每个Application都有自己的Executor，与其他程序完全隔离
2.2.Spark-2.3.4 standlone搭建
1.获取
wget http://mirrors.tuna.tsinghua.edu.cn/apache/spark/spark-2.3.4/spark-2.3.4-bin-hadoop2.7.tgz
2.解压
tar -xf spark-2.3.4-bin-hadoop2.7.tgz
3.移动到/opt/bigdata中
mv spark-2.3.4-bin-hadoop2.7.tgz /opt/bigdata
4.修改环境变量SPARK_HOME
这步省略（如果不知道如何修改环境变量，参考我的Hadoop Ha搭建文章）
5.进入SPARK_HOME/bin/spark-shell，我们进入一个scala交互环境
这边可以测试wc代码，默认sparkContext自动为sc，所以wc代码如下
sc.textFile("/root/data.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).foreach(println);
2.3.Spark集群搭建
1.修改conf/spark-env.sh
export HADOOP_CONF_DIR=/opt/bigdata/hadoop-2.6.5
为了找到我们之前配置好的hadoop集群
export SPARK_MASTER_HOST=node01
export SPARK_MASTER_PORT=7077
export SPARK_MASTER_WEBUI_PORT=8080
export SPARK_WORKER_CORES=2
export SPARK_WORKER_MEMORY=1g
2.修改slaves
node02
node03
node04
拷贝这两个配置到其他三个节点
3.spark-shell集群式启动
start-all.sh
spark-shell --master spark://node01:7077
进入scala的交互环境，运行一下wc
sc.textFile("hdfs://mycluster/sparktest/data.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).foreach(println)
发现没有输入，因为在分区上输出了，所以我们需要一个回收算子
sc.textFile("hdfs://mycluster/sparktest/data.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect().foreach(println)
2.3.Spark高可用处理
1.spark-defaults.conf修改
这边的配置可以向zookeeper上面进行注册发现
spark.deploy.recoveryMode ZOOKEEPER
spark.deploy.zookeeper.url node02:2181,node03:2181,node04:2181
spark.deploy.zookeeper.dir /littlepagespark
2.启动流程
./spark-shell --master spark://node01:7077,node02:7077
提示：node02的master需要自行启动，所以我们需要更改node02的conf/env文件
三、history服务
1.配置spark-defaults.conf
spark.eventLog.enabled true
spark.eventLog.dir hdfs://mycluster/shared/spark-logs
spark.history.fs.logDirectory hdfs://mycluster/spark_log
手动开启历史服务./start-history-server.sh
之后访问node01:18080端口即可
四、使用spark-submit进行计算Pi
代码参照官网
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// scalastyle:off println
package org.apache.spark.examples
import scala.math.random
import org.apache.spark.sql.SparkSession
/** Computes an approximation to pi */
object SparkPi {
def main(args: Array[String]): Unit = {
val spark = SparkSession
.builder
.appName("Spark Pi")
.getOrCreate()//创建一个spark pi
//如果参数大于0，则返回切片数量为args,否则返回2
val slices = if (args.length > 0) args(0).toInt else 2
//n为切片乘100000L和Int最大值的最大，转为Int，避免溢出
val n = math.min(100000L * slices, Int.MaxValue).toInt // avoid overflow
//计算随机数，是否落到圆上
val count = spark.sparkContext.parallelize(1 until n, slices).map { i =>
val x = random * 2 - 1
val y = random * 2 - 1
if (x*x + y*y <= 1) 1 else 0
}.reduce(_ + _)
//打印
println(s"Pi is roughly ${4.0 * count / (n - 1)}")
spark.stop()
}
}
// scalastyle:on println
使用Spark-submit运行
spark-submit --master spark://node01:7077,node02:7077 --class org.apache.spark.examples.SparkPi ./spark-examples_2.11-2.3.4.jar 1000
spark-submit --master url --class [package.class] [jar] [args]
spark-submit参数
--deploy-mode cluster #部署模式，默认cluster
--driver-memory 1024m #driver内存
--total-executor-cores 6 #总共6核心
--executor-cores 1 #每个executor1个核
--executor-memory 1024m #executor内存1024m
五、Spark On Yarn
在hadoop-env.sh仅需要配置export hadoop_conf_dir的目录即可，worker和master均不需要，slaves也不需要了，zk的配置也不需要了
1.hadoop-env.sh修改
export HADOOP_CONF_DIR=/opt/bigdata/hadoop-2.6.5/etc/hadoop
2.spark-defaults.conf修改
spark.eventLog.enabled true
spark.eventLog.dir hdfs://namenode/shared/spark-logs
spark.history.fs.logDirectory hdfs://mycluster/spark_log
3.yarn-site.xml
<property>
<name>yarn.nodemanager.resource.memory-mb</name>
<value>4096</value>
</property>
<property>
<name>yarn.nodemanager.resource.cpu-vcores</name>
<value>4</value>
</property>
<property>
<name>yarn.nodemanager.vmem-check-enabled</name>
<value>false</value>
</property>
4.mapred-site.xml
mr-jobhistory-daemon.sh start historyserver
<property>
<name>mapred.job.history.server.embedded</name>
<value>true</value>
</property>
<property>
<name>mapreduce.jobhistory.address</name>
<value>node03:10020</value>
</property>
<property>
<name>mapreduce.jobhistory.webapp.address</name>
<value>node03:50060</value>
</property>
<property>
<name>mapreduce.jobhistory.intermediate-done-dir</name>
<value>/work/mr_history_tmp</value>
</property>
<property>
<name>mapreduce.jobhistory.done-dir</name>
<value>/work/mr-history_done</value>
</property>
启动历史服务
mr-jobhistory-daemon.sh start historyserver
启动spark-yarn
./spark-shell --master yarn
六、shell脚本
class=org.apache.spark.examples.SparkPi
jar=$SPARK_HOME/examples/jars/spark-examples_2.11-2.3.4.jar
master=yarn
$SPARK_HOME/bin/spark-submit
--master $master
--class $class
--deploy-mode cluster
$jar
100
七、调优
spark-defaults.conf增加配置
spark.yarn.jars hdfs://mycluster/work/spark_lib/jars/*
把spark jars目录完全上传
hdfs dfs -put ./* /work/spark_lib/jars/
其余调优详见四的参数