一、WordCount示例

// 基本实现
package com.pg.learning.demo
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object FirstSparkDemo {
  def main(args: Array[String]): Unit = {
    //通过SparkContext建立与spark之间的连接，设置对应的属性
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    val sc = new SparkContext(sparkConf)
    //通过连接进行操作的业务
    val line: RDD[String] = sc.textFile("words.txt")
    //val line: RDD[String] = sc.wholeTextFile(path , minPartitions=2)// 读取小文件可用，通过设置minPartitions避免过多的分区
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = line.flatMap(_.split(" ")).map((_,1)).groupBy(_._1)
    val result: RDD[(String, Int)] = groupRDD.map{case(word,list) => {(word,list.size)}}
    //输出
    val tuples: Array[(String, Int)] = result.collect()
    tuples.foreach(println)
    //关闭连接
    sc.stop()
  }
}
//使用spark的api实现
object WordCountWithSparkAPI {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCountWithSparkAPI")
    val sc = new SparkContext(sparkConf)
    val line: RDD[String] = sc.textFile("words.txt")
    val mapRDD: RDD[(String, Int)] = line.flatMap(_.split(" ")).map((_,1))
    //reduceByKey:相同key的数据，可以对value进行reduce聚合
    val result: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)
    val tuples: Array[(String, Int)] = result.collect()
    tuples.foreach(println)
    sc.stop()
  }
}

二、spark运行模式

2.1、local模式

1、通过命令进行命令行模式：/bin/spark-shell ，可以在命令行工具执行spark语句。
2、通过sumbit提交：

spark-submit \
--master local[2] \
--class T1 \
--name T1 \
--driver-memory 2g \
--driver-cores 2 \
--num-executors 10 \
--executor-cores 1 \
--executor-memory 2g \
--conf spark.yarn.executor.memoryOverhead=4096 \
--conf spark.network.timeout=10000000 \
--conf spark.executor.heartbeatInterval=10000000 \
--conf spark.shuffle.service.enabled=true \
--conf "spark.driver.extraJavaOptions=-XX:MaxPermSize=1g -XX:+UseConcMarkSweepGC" \
--conf "spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC" \
/data/batchJob/XXXHdfs.jar
## 通过spark-shell --help 可以查看帮助参数

—master local：表示本地运行，只有一个工作进程，无并行计算能力。
—master local[K]：表示本地运行，有K个工作进程，通常K设置为CPU的核数。
—master local[*]：表示本地运行，工作进程数等于机器的CPU核数数量。[

](https://blog.csdn.net/silentanytime/article/details/107035491)

2.2、Standlone模式

独立部署（Standalone）模式是Spark 自身节点运行的集群模式，Spark 的Standalone 模式体现了经典的master-slave 模式。集群一般规划如下：

	Linux1	Linux2	Linux3
Spark	Worker Master	Worker	Worker

具体安装步骤如下：

1、解压缩文件
将 spark-3.0.0-bin-hadoop3.2.tgz 文件上传到Linux 并解压缩在指定位置
tar -zxvf spark-3.0.0-bin-hadoop3.2.tgz -C /opt/module cd /opt/module
mv spark-3.0.0-bin-hadoop3.2 spark-standalone
2、修改配置文件
1)进入解压缩后路径的 conf 目录，修改 slaves.template 文件名为 slaves
mv slaves.template slaves    
2)修改 slaves 文件，添加work节点，内容为如下：
linux1
linux2
linux3
3)修改 spark-env.sh.template 文件名为 spark-env.sh
mv spark-env.sh.template spark-env.sh    
4)修改 spark-env.sh 文件，添加 JAVA_HOME 环境变量和集群对应的 master 节点，内容为：
export JAVA_HOME=/opt/module/jdk1.8.0_144 
SPARK_MASTER_HOST=linux1 # 指定master的ip
SPARK_MASTER_PORT=7077   # 指定提交任务的端口
SPARK_MASTER_WEBUI_PORT=8080
HADOOP_CONF_DIR=...  #读取hadoop配置的路径
YARN_CONF_DIR=...    #读取yarn配置的路径
注意：7077 端口，相当于 hadoop3 内部通信的 8020 端口，此处的端口需要确认自己的Hadoop配置
5)分发 spark-standalone 目录
xsync spark-standalone    
3、启动集群
1)执行脚本命令：
sbin/start-all.sh
4、可以查看三台服务器的进程
================linux1================ 
3330 Jps
3238 Worker
3163 Master
================linux2================ 
2966 Jps
2908 Worker
================linux3================
2978 Worker
3036 Jps
5、查看 Master 资源监控Web UI 界面: http://linux1:8080
6、提交应用命令：
bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://linux1:7077 \
./examples/jars/spark-examples_2.12-3.0.0.jar \ 10
参数说明：
1)--class 表示要执行程序的主类
2)--master spark://linux1:7077 独立部署模式，连接到Spark 集群，一般有模式：local[*]、spark://linux1:7077、Yarn
3)spark-examples_2.12-3.0.0.jar 运行类所在的 jar 包，包含依赖。这个URL在集群中全局可见。比如 hdfs://共享存储系统，如果是file:// path， 那么所有的节点的
path 都包含同样的 jar
4)数字 10 表示程序的入口参数，用于设定当前应用的任务数量
5)--executor-memory 1G：指定每个executor可用内存为 1G
6)--total-executor-cores 2：指定所有executor使用的cpu 核数为2个
7)--executor-cores：指定每个executor 使用的cpu 核数
8)application-arguments：传给 main()方法的参数
7、执行任务时，会产生多个 Java 进程：
一般会有：SparkSubmit、CoarseGrainedExecutorBackend

2.3、配置历史服务器

由于 spark-shell 停止掉后，集群监控 linux1:4040 页面就看不到历史任务的运行情况，所以开发时都配置历史服务器记录任务运行情况。

1)修改 spark-defaults.conf.template 文件名为 spark-defaults.conf
mv spark-defaults.conf.template spark-defaults.conf
2)修改 spark-default.conf 文件，配置日志存储路径
spark.eventLog.enabled true
spark.eventLog.dir hdfs://linux1:8020/directory
注意：需要启动 hadoop 集群，HDFS 上的directory 目录需要提前存在。
sbin/start-dfs.sh
hadoop fs -mkdir /directory
3)修改 spark-env.sh 文件, 添加日志配置
export SPARK_HISTORY_OPTS="
-Dspark.history.ui.port=18080
-Dspark.history.fs.logDirectory=hdfs://linux1:8020/directory
-Dspark.history.retainedApplications=30"
⚫参数 1 含义：WEB UI 访问的端口号为 18080
⚫参数 2 含义：指定历史服务器日志存储路径
⚫参数 3 含义：指定保存Application 历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上显示的应用数。
3.1)如果是yarn模式的还需要修改配置spark-defaults.conf
spark.yarn.historyServer.address=linux1:18080
spark.history.ui.port=18080
4)分发配置文件
5)重新启动集群和历史服务
sbin/start-all.sh
sbin/start-history-server.sh
6)重新执行任务，就可以在：http://linux1:18080，查看历史服务。

2.4、配置高可用（ HA）

所谓的高可用是因为当前集群中的 Master 节点只有一个，所以会存在单点故障问题。所以为了解决单点故障问题，需要在集群中配置多个 Master 节点，一旦处于活动状态的 Master 发生故障时，由备用 Master 提供服务，保证作业可以继续执行。这里的高可用一般采用Zookeeper 设置，zookeeper配置如下：

	Linux1	Linux2	Linux3
Spark	Master Zookeeper Worker	Master Zookeeper Worker	Zookeeper Worker

1)停止集群
sbin/stop-all.sh    
2)启动Zookeeper
xstart zk    
3)修改 spark-env.sh 文件添加如下配置
注 释 如 下 内 容 ： 
#SPARK_MASTER_HOST=linux1 
#SPARK_MASTER_PORT=7077
添加如下内容:
#Master 监控页面默认访问端口为 8080，但是可能会和 Zookeeper 冲突，所以改成 8989，也可以自定义，访问 UI 监控页面时请注意
SPARK_MASTER_WEBUI_PORT=8989
export SPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER
-Dspark.deploy.zookeeper.url=linux1:2181,linux2:2181,linux3:2181
-Dspark.deploy.zookeeper.dir=/spark"
4)分发配置文件
xsync conf/    
5)启动集群
sbin/start-all.sh
6)启动 linux2 的单独 Master 节点，此时 linux2 节点 Master 状态处于备用状态
sbin/start-master.sh
7)提交应用到高可用集群
bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://linux1:7077,linux2:7077 \
./examples/jars/spark-examples_2.12-3.0.0.jar \ 10

2.5、Yarn模式

独立部署（Standalone）模式由 Spark 自身提供计算资源，无需其他框架提供资源。这种方式降低了和其他第三方资源框架的耦合性，独立性非常强。但是你也要记住，Spark 主要是计算框架，而不是资源调度框架，所以本身提供的资源调度并不是它的强项，所以还是和其他专业的资源调度框架集成会更靠谱一些。
SparkOnYarn根据Driver运行在哪里可以区分两种模式：client模式和cluster模式。

1)修改 hadoop 配置文件/opt/module/hadoop/etc/hadoop/yarn-site.xml
<!--是否启动一个线程检查每个任务正使用的物理内存量，如果任务超出分配值，则直接将其杀掉，默认是 true -->
<property>
<name>yarn.nodemanager.pmem-check-enabled</name>
<value>false</value>
</property>
<!--是否启动一个线程检查每个任务正使用的虚拟内存量，如果任务超出分配值，则直接将其杀掉，默认是 true -->
<property>
<name>yarn.nodemanager.vmem-check-enabled</name>
<value>false</value>
</property>
2)修改 conf/spark-env.sh，添加 JAVA_HOME 和YARN_CONF_DIR 配置
mv spark-env.sh.template spark-env.sh
export JAVA_HOME=/opt/module/jdk1.8.0_144 
YARN_CONF_DIR=/opt/module/hadoop/etc/hadoop
HADOOP_CONF_DIR=/opt/module/hadoop/etc/hadoop #需要使用hdfs时配置
3)启动yarn和hdfs
4)启动spark
5)提交作业
bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode cluster \
./examples/jars/spark-examples_2.12-3.0.0.jar \ 10
master 需要指定为yarn；
deploy-mode 可以指定为cluster模式或者client模式，client模式会把结果打印在控制台，而cluster模式不会。

2.5.1、client模式

client模式下：spark的Driver驱动程序，运行在提交任务的客户端上，和集群的通讯成本高。因为Driver在客户端，所以在Driver中的程序结果输出可以在客户端的控制台看到。

2.5.2、cluster模式

cluster模式下：Spark的Driver驱动程序，运行在yarn集群上，和集群的通讯成本低。且Driver是交给Yarn管理的，如果失败了会由yarn重启，因为Driver运行在yarn上，所以本地客户端控制台看不到输出的结果，需要在Yarn日志中查看。

2.6、部署模式对比

模式	Spark 安装机器数	需启动的进程	所属者	应用场景
Local	1	无	Spark	测试
Standalone	3	Master 及Worker	Spark	单独部署
Yarn	1	Yarn 及HDFS	Hadoop	混合部署

2.7、常用端口号

Ø Spark 查看当前Spark-shell 运行任务情况端口号：4040（计算）
Ø Spark Master 内部通信服务端口号：7077
Ø Standalone 模式下，Spark Master Web 端口号：8080（资源）
Ø Spark 历史服务器端口号：18080
Ø Hadoop YARN 任务运行情况查看端口号：8088

Spark学习笔记

1、Spark入门