3 Spark 应用架构
登录到Spark HistoryServer历史服务器WEB UI界面，点击刚刚运行圆周率PI程序：
查看应用运行状况：
切换到【Executors】Tab页面：
从图中可以看到Spark Application运行到集群上时，由两部分组成：Driver Program和Executors。
- 第一、Driver Program
相当于AppMaster，整个应用管理者，负责应用中所有Job的调度执行;
运行JVM Process，运行程序的MAIN函数，必须创建SparkContext上下文对象；
一个SparkApplication仅有一个；
第二、Executors
相当于一个线程池，运行JVM Process，其中有很多线程，每个线程运行一个Task任务，
一个Task运行需要1 Core CPU，所有可以认为Executor中线程数就等于CPU Core核数；一个Spark Application可以有多个，可以设置个数和资源信息；
Driver Program是用户编写的数据处理逻辑，这个逻辑中包含用户创建的SparkContext。
SparkContext 是用户逻辑与Spark集群主要的交互接口，它会和Cluster Manager交互，包括向它
申请计算资源等。 Cluster Manager负责集群的资源管理和调度，现在支持Standalone、Apache
Mesos和Hadoop的 YARN。Worker Node是集群中可以执行计算任务的节点。 Executor是在一
个Worker Node上为某应用启动的一个进程，该进程负责运行任务，并且负责将数据存在内存或者
磁盘上。Task 是被送到某个Executor上的计算单元，每个应用都有各自独立的 Executor，计算最终在计算节点的 Executor中执行。
用户程序从最开始的提交到最终的计算执行，需要经历以下几个阶段：
1）、用户程序创建 SparkContext 时，新创建的 SparkContext 实例会连接到 ClusterManager。
Cluster Manager 会根据用户提交时设置的 CPU 和内存等信息为本次提交分配计算资源，启
动 Executor。
2）、Driver会将用户程序划分为不同的执行阶段Stage，每个执行阶段Stage由一组完全相同Task
组成，这些Task分别作用于待处理数据的不同分区。在阶段划分完成和Task创建后， Driver
会向Executor发送 Task；
3）、Executor在接收到Task后，会下载Task的运行时依赖，在准备好Task的执行环境后，会开
始执行Task，并且将Task的运行状态汇报给Driver；
4）、Driver会根据收到的Task的运行状态来处理不同的状态更新。 Task分为两种：一种是Shuffle
Map Task，它实现数据的重新洗牌，洗牌的结果保存到Executor 所在节点的文件系统中；另
外一种是Result Task，它负责生成结果数据；
5）、Driver 会不断地调用Task，将Task发送到Executor执行，在所有的Task 都正确执行或者
超过执行次数的限制仍然没有执行成功时停止；
4 WEB UI 监控
Spark 提供了多个监控界面，当运行Spark任务后可以直接在网页对各种信息进行监控查看。
运行spark-shell交互式命令在Standalone集群上，命令如下：
/export/server/spark/bin/spark-shell --master spark://node1.oldlu.cn:7077
运行截图如下所示：
在node3.oldlu.cn运行spark-shell，WEB UI监控页面地址：http://node3.oldlu.cn:4040
在spark-shell中执行词频统计WordCount程序代码，运行如下：
val inputRDD = sc.textFile("/datas/wordcount.data") val wordcountsRDD = inputRDD.flatMap(line => line.split("\\s+")).map(word => (word, 1)).reduceByKey((tmp, item) => tmp +item) wordcountsRDD.take(5)
截图如下：
可以发现在一个Spark Application中，包含多个Job，每个Job有多个Stage组成，每个Job执行
按照DAG图进行的。
其中每个Stage中包含多个Task任务，每个Task以线程Thread方式执行，需要1Core CPU。
可以看到Spark为应用程序提供了非常详尽的统计页面，每个应用的Job和Stage等信息都可以
在这里查看到。通过观察应用详情页的各个信息，对进一步优化程序，调整瓶颈有着重要作用，后
期综合项目案例详细讲解。
Spark Application程序运行时三个核心概念：Job、Stage、Task，说明如下：
Task：被分配到各个 Executor 的单位工作内容，它是 Spark 中的最小执行单位，一
般来说有多少个 Paritition（物理层面的概念，即分支可以理解为将数据划分成不同
部分并行处理），就会有多少个 Task，每个 Task 只会处理单一分支上的数据。
Job：由多个 Task 的并行计算部分，一般 Spark 中的 action 操作（如 save、collect，后面
进一步说明），会生成一个 Job。
Stage：Job 的组成单位，一个 Job 会切分成多个 Stage，Stage 彼此之间相互依赖顺序执行，
而每个 Stage 是多个 Task 的集合，类似 map 和 reduce stage。
5 Standalone HA
Spark Standalone集群是Master-Slaves架构的集群模式，和大部分的Master-Slaves结构集群
一样，存在着Master单点故障（SPOF）的问题。
5.1 高可用HA
如何解决这个单点故障的问题，Spark提供了两种方案：
-基于文件系统的单点恢复(Single-Node Recovery with Local File System)；
-基于Zookeeper的Standby Masters(Standby Masters with ZooKeeper)；
ZooKeeper提供了一个Leader Election机制，利用这个机制可以保证虽然集群存在多个
Master，但是只有一个是Active的，其他的都是Standby。当Active的Master出现故障时，另外的
一个Standby Master会被选举出来。由于集群的信息，包括Worker， Driver和Application的信息
都已经持久化到文件系统，因此在切换的过程中只会影响新Job的提交，对于正在进行的Job没有任
何的影响。加入ZooKeeper的集群整体架构如下图所示。
5.2 基于Zookeeper实现HA
官方文档： http://spark.apache.org/docs/2.4.5/spark-standalone.html#standby-masters-with-zookeeper
- 1）、停止Standalone集群
## 在node1.oldlu.cn上执行命令 /export/server/spark/sbin/stop-master.sh /export/server/spark/sbin/stop-slaves.sh
- 2）、增加Zookeeper配置
对Spark配置文件【$SPARK_HOME/conf/spark-env.sh】文件如下修改：
SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=node1.oldlu.cn:2181,node2.oldlu.cn:2181,node3.oldlu.cn:2181 -Dspark.deploy.zookeeper.dir=/spark-ha"
参数含义说明：
spark.deploy.recoveryMode：恢复模式 spark.deploy.zookeeper.url：ZooKeeper的Server地址 spark.deploy.zookeeper.dir：保存集群元数据信息的文件、目录。包括Worker、Driver、Application信息。
注释或删除MASTER_HOST内容：
# SPARK_MASTER_HOST=node1.oldlu.cn
- 3）、将spark-env.sh分发集群
cd /export/server/spark/conf scp -r spark-env.sh root@node2.oldlu.cn:$PWD scp -r spark-env.sh root@node3.oldlu.cn:$PWD
- 4）、启动集群服务
先启动Zookeeper集群，再分别启动2个Master服务，最后启动Worker服务
## 启动ZOOKEEPER服务 zookeeper-daemons.sh start ## 在node1和node2分别启动Master服务 /export/server/spark/sbin/start-master.sh ## 查看哪个Master为Active，就在哪个Master机器上启动Workers服务 /export/server/spark/sbin/start-slaves.sh
默认情况下，先启动Master就为Active Master，如下截图所示：
5.3 测试运行
Standalone HA集群运行应用时，指定ClusterManager参数属性为
--master spark://host1:port1,host2:port2
提交圆周率PI运行集群，命令如下
SPARK_HOME=/export/server/spark ${SPARK_HOME}/bin/spark-submit \ --master spark://node1.oldlu.cn:7077,node2.oldlu.cn:7077 \ --class org.apache.spark.examples.SparkPi \ ${SPARK_HOME}/examples/jars/spark-examples_2.11-2.4.5.jar \ 100
在执行过程中，使用jps查看Active Master进程ID，将其kill，观察Master是否自动切换与应用运
行完成结束。
