SparkCore学习笔记（一）

106 阅读 0 评论 70 点赞

我是靠谱客的博主傻傻毛巾，最近开发中收集的这篇文章主要介绍SparkCore学习笔记（一），觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

2017/1/10 8:27

Spark架构原理

**概念词汇：**Application，Application Jar，Driver program（驱动程序）、Cluster Manager（管理集群资源的外部服务）、Deploy mode（决定在何处运行Driver进程的部署模式：cluster和client）、worker node、Executor、Task、job、stage（阶段）、RDD、Operation（transformation和action）、partition、DAG、Narrow dependency、Wide dependency、CaChing Management

Standalone集群模式

1.Driver进程：

2.Master进程：负责资源的调度和分配、集群的监控。

3.worker进程：主要职责

存储RDD的partition
启动其他的进程和线程，对RDD进行并行计算处理

4.Excutor：启动线程来执行task

5.task线程：对RDD的partition数据执行指定的算子操作

client提交应用，master找到worker启动Driver，Driver向master申请资源，之后将其转化为RDDGraph
再由DAGscheduler将其转化为statge的有向无环图交给taskScheduler，
TaskScheduler将任务交给executor执行。

架构原理简述

Client上提交应用请求给master
MAster在接受请求后开始进行资源调度（集群资源分配算法），调用worker资源。【Executor的分配】
worker在接收到master的请求后，会为spark应用启动Executor。
Executor启动后，会向Driver进行反注册，这样Driver就可以知道那些Excutor在为他继续服务了。
Driver在注册一些Executor后，就可以开始执行应用了，首先初始化RDD，告知Master，并进行注册，启动程序。
HDFS的文件被读取到多个worker节点上，形成分布式数据集，也就是初始RDD（读数据的过程）。
Driver根据我们对RDD的操作，提交一堆task到Executor上。Executor接收到task后，启动多个线程来执行task。
task会对RDD的partition数据进行指定的算子操作，形成新的RDD的partition。

YARN模式

运行原理：

（yarn-cluster提交模式）

spark-submit提交，将请求发送到ResourceManager，请求启动ApplicationMaster。
RM对资源进行调度分配，分配container在某些nodeManager上，启动ApplicationMaster
ApplicationMaster（相当于Driver）会找RM，请求container启动Executor。
RM分配一批container用于启动Executor，AM链接相应的NM来启动Executor，（NodeManager相当于之前的Worker）。
Executor在启动后，向AM反向注册，告诉AM哪些节点参与了工作。
读数据、计算、产生结果

（yarn-client模式）

Driver进程是在本地启动的。用户提交请求给ResourcesManager（RM），请求启动ApplicationMaster（AM）。
RM进行集群资源分配，分配container在某个NodeMannager上，（AM在这里起ExecutorLauncher的作用）
AM申请Container。启动Executor。
RM分配给一批container，然后AM链接其他NM，用container的资源启动Executor
Executor反省注册到本地的Driver上。
初始化ＲＤＤ，读取数据，执行Ｔａｓｋ，产生结果。

比较：yarn-clusteryarn-client:

yarn-client用于测试，因为Driver运行在本地客户端。会和yarn集群进行通讯，会产生大量流量，直接运行的时候，本地可以看到所有log
yarn-cluster用于生产环境，Driver运行在nodemanager。产生的日志会分散到各个节点上，不方便查看。

分组和二次排序

排序：（初级版）

import org.apache.spark.{SparkConf, SparkContext}
/**
* Created by yxy on 1/9/17.
*/
object sortDemo {
def main(args: Array[String]): Unit = {
sortedDemo
}
def sortedDemo: Unit ={
val conf=new SparkConf().setAppName("sortDemo").setMaster("local")
val sc=new SparkContext(conf)
val infoRDD=sc.textFile("hdfs://hadoop01:9000/data/student.txt")
val result=infoRDD.map(_.split(" "))
.filter(m=>m.length==2 && m(0)!="" && m(1)!=" ") //记得判断分割后的数据是否满足格式哦！
.map(m=>{
val classinfo=m(0).trim
val score=m(1).trim.toInt
(classinfo,score)
})
.groupByKey()
.map(m=>{
val classname=m._1
val scores=m._2.toArray.sortWith(_>_).take(3)
(classname,scores)
})
.foreach(m=>{
println(m._1+" : ")
for (elem <- m._2) {
println("score :"+elem)
}
})
}
}

（待续）