Spark集群
0.0体验安装Spark在集群单节点
1.tar
tar -xzvf xxx.tgz -C /soft/
ln -s /soft/spark-2.1.0-bin-hadoop2.7 /soft/spark
2.配置环境变量
SPARK_HOME=/soft/spark
PATH=$SPARK_HOME/bin:$SPARK_HOME/sbin:$PATH
3.启动spark shell
RDD:
Resilient Distributed dataset , 弹性分布式数据集。List
使用spark实现wordcount:
//分步实现
val rdd1 = sc.textFile("file:///home/centos/1.txt");
val rdd2 = rdd1.flatMap(line=>{line.split(" ")})
val rdd3 = rdd2.map(word=>{(word , 1)})
val rdd4 = rdd3.reduceByKey((a:Int,b:Int)=>{a + b})
rdd4.collect()
//一步完成(reduceByKey)
sc.textFile("file:///home/centos/1.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect()
//一步完成(groupByKey)
sc.textFile("file:///home/centos/1.txt").flatMap(_.split(" ")).map((_,1)).groupByKey().mapValues(_.size).collect()
===========================================
Spark集群分类
1.local
使用一个jvm模拟spark集群
2.standalone
启动master + worker进程
3.mesos
--
4.yarn
spark on yarn , 在yarn执行spark的job。根本不需要spark集群。
spark只是一个hadoop的job。
2."standalone"模式spark集群
2.1（安装+standalone）
1.规划
s201 ~ s204
s201 //master(进程)
s202 ~ s204 //worker(进程)
2.在每个主机上分别安装spark
3.配置spark(每个节点进行相同操作)
3.1)在spark的conf目录下创建hadoop的core-site.xml和hdfs-site.xml软连接
$>xcall.sh ln -s /soft/hadoop/etc/hadoop/core-site.xml /soft/spark/conf/core-site.xml
$>xcall.sh ln -s /soft/hadoop/etc/hadoop/hdfs-site.xml /soft/spark/conf/hdfs-site.xml
3.2)配置JAVA_HOME环境变量
$>cd /soft/spark/conf
$>cp spark-env.sh.template spark-env.sh
$>nano spark-env.sh
...
export JAVA_HOME=/soft/jdk
...
3.3)修改slaves文件
$>cd /soft/spark/conf
$>cp slaves.template slaves
$>nano slaves
s202
s203
s204
3.4)分发配置文件
$>xsync.sh spark-env.sh
$>xsync.sh slaves
4.启动spark集群
4.1)启动hadoop集群
xzk.sh start //启动zk
start-dfs.sh //启动hdfs
4.2)启动spark（standalone模式）
$>cd /soft/spark/sbin
//
$>./start-all.sh
5.webui
http://s201:8080
6.启动spark shell，连接到spark集群
$>spark-shell --master spark://s201:7077
$>sc.textFile("/user/centos/data/1.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect()
2.2导出idea下的项目成为jar包，部署到spark集群上运行
1.修正代码
if (args == null || args.length == 0) {
throw new Exception("需要指定文件路径") ;
}
SparkConf conf = new SparkConf();
conf.setAppName("tagTenJava");
//conf.setMaster("local");
2.导出jar，添加依赖的第三方类库
导入第三方依赖的库.
3.传输到centos
4.使用spark-submit提交应用
spark-submit --class TaggenScala1 --master spark://s201:7077 myspark.jar /user/centos/data/tags.txt
spark-submit --class com.p2p.spark.mr.TaggenJava2 --master spark://s201:7077 myspark.jar /user/centos/data/tags.txt
spark下运行job，使用第三方的jar的方式
1.将第三方jar分发到所有的spark/jars下
2.将第三方jar打散，和我们自己的jar打到一起。
3.在spark-submit命令中，通过--jars指定使用的第三方jar包
使用spark-shell执行taggen
1.启动spark-shell，指定fastjson类库。
定位到fastjson jar文件
C:UsersAdministrator.m2
epositorycomalibabafastjson1.2.24fastjson-1.2.24.jar
2.启动spark-shell
$>spark-shell --master spark://s201:7077 --jars /home/centos/fastjson-1.2.24.jar
3.查看webui
http://s201:8080/
4.
5.
Spark job部署模式 --deploy-mode client VS cluster
spark job的部署有两种模式，client和cluster
spark-submit .. --deploy-mode client | cluster
1.client
默认值，driver运行在client端主机上。
spark-submit --class com.p2p.spark.mr.TaggenJava2 --master spark://s201:7077 hdfs://s201/user/centos/data/myspark.jar /user/centos/data/tags.txt
2.cluster
driver运行在某个worker节点上。客户端值负责提交job。
$>spark-submit --class com.p2p.spark.mr.TaggenJava2 --master spark://s201:7077 --deploy-mode cluster hdfs://s201/user/centos/data/myspark.jar /user/centos/data/tags.txt
3.spark集群模式之yarn
3.1部署集群：
1.停止spark集群
2.启动yarn集群
3.配置HADOOP_CONF_DIR环境变量
[spark-env.sh]
...
export HADOOP_CONF_DIR=/soft/hadoop/etc/hadoop
4.启动shell,使用yarn的方式启动
spark-shell --master yarn --num-executors 4 --executor-cores 5 --executor- memory 1g
5.配置spark归档文件，避免每次spark on yarn重新上传spark资源文件。
5.1)上传zip文件到hdfs
hdfs dfs -put __spark_libs__385163057195536794.zip data
5.2)配置conf/spark-defaults.conf
spark.yarn.archive hdfs://mycluster/user/centos/data/__spark_libs.zip
#spark.dynamicAllocation.enabled true
#spark.shuffle.service.enabled true
3.2故障
1.虚拟内存超限
[yarn-site.xml]
<property>
<name>yarn.nodemanager.vmem-check-enabled</name>
<value>false</value>
</property>
3.3使用spark on yarn client 模式运行气温数据
spark-submit --master yarn --deploy-mode client --class TempAggDemoScala_GroupByKey myspark.jar
3.4sparktask、线程、executor之间的关系
spark.task.cpus 2 //每个task分配的core数，默认1
--executor-cores 5 //每个执行器占用的core数.
--num-executors 2 //yarn模式指定执行器个数
solt //槽位，本质上对应一个线程，slot数指job最大的并发度。
//最发并发度是job同时启动的线程数。 = executors个数 * 每个执行内核数 / 每个task占用内核数。
-- 配置方式 , 开启4个线程，
spark-shell --master yarn --deploy-mode client --executor-cores 5 --num-executors 2 --conf spark.task.cpus=2
$scala>sc.makeRDD(1 to 5 , 5).map(e=>{sendInfo(this , "map" , e+ "") ; e * 2}).collect