大数据-----spark底层通信交互简单原理与实现
1.spark是什么？
Spark是一个基于内存的大数据计算引擎。提高了在大数据环境下数据处理的实时性，spark仅仅涉及到了数据的计算，没有涉及到数据的存储。
Spark是由Scala语言编写的，
2.什么是scala
Scala 是一种多范式的编程语言，其设计的初衷是要集成面向对象编程和函数式编程的各种特性。Scala 运行于 Java 平台（Java 虚拟机），并兼容现有的Java 程序
3.Scala的特点：
3.1 速度快：Scala 语言表达能力强，一行代码抵得上 Java 多行，开发速度快；Scala 是静态编译的，所以和 JRuby,Groovy 比起来速度会快很多
3.2 能融合到 Hadoop 生态圈：Hadoop 现在是大数据事实标准，Spark 并不是要取代 Hadoop，而是要完善 Hadoop 生态。JVM 语言大部分可能会想到 Java，但Java 做出来的 API 太丑，或者想实现一个优雅的 API 太费劲。
4.spark与scala
Spark是用Scala语言编写的，如果想应用spark这个技术了解更多的底层逻辑原理，就要回Scala这门编程语言，Scala的程序运行需要 JVM 平台，所以安装 Scala 之前要安装 JDK，Scala的程序逻辑与java很相似，增加了函数式编程，特殊情况有所不同，代码格式也有所不同。对于Scala基础这里不做多的说明。
5.Scala actor
Scala 中的 Actor 能够实现并行编程的强大功能，它是基于事件模型的并发机制，Scala
是运用消息的发送、接收来实现高并发的，Actor 可以看作是一个个独立的实体，他们之间是毫无关联的。但是，他们可以通过消息来通信。一个 Actor 收到其他 Actor 的信息后，它可以根据需要作出各种相应。消息的类型可以是任意的，消息的内容也可以是任意的（Scala Actor 是 scala 2.10.x 版本及以前版本的 Actor。Scala 在 2.11.x 版本中将 Akka 加入其中，作为其默认的 Actor，老版本的 Actor 已经废弃）akka是以actor为模型的，学习actor是为了akka做准备。
Actor的执行顺序：
1）、首先调用 start()方法启动 Actor
2）、调用 start()方法后其 act()方法会被执行
3）、向 Actor 发送消息
4）、act 方法执行完成之后，程序会调用 exit 方法
6.akka
Spark 的 RPC 是通过 Akka 类库实现的，Akka 用 Scala 语言开发，基于 Actor并发模型实现，Akka 具有高可靠、高性能、可扩展等特点，使用 Akka 可以轻松实现分布式 RPC 功能
Akka 基于 Actor 模型，提供了一个用于构建可扩展的（Scalable）、弹性的（Resilient）、快速响应的（Responsive）应用程序的平台。Actor 模型：在计算机科学领域，Actor 模型是一个并行计算（ConcurrentComputation）模型，它把 actor 作为并行计算的基本元素来对待：为响应一个接收到的消息，一个 actor 能够自己做出一些决策，如创建更多的 actor，或发
送更多的消息，或者确定如何去响应接收到的下一个消息。
Actor是Akka中最核心的概念，它是一个封装了状态和行为的对象，Actor之间可以通过交换消息的方式进行通信，每个Actor都有自己的收件箱（Mailbox）。通过Actor能够简化锁及线程管理，可以非常容易地开发出正确地并发程序和并行系统，Actor具有如下特性
(1)、提供了一种高级抽象，能够简化在并发（Concurrency）/并行（Parallelism）应用场景下的编程开发
（2）、提供了异步非阻塞的、高性能的事件驱动编程模型
（3）、超级轻量级事件处理（每GB堆内存几百万Actor）
7.使用Akka实现一个简易版的spark通信框架：
7.1 架构图：
7.2 代码实现：
7.2.1Master类
//todo:利用akka实现简易版的spark通信框架-----Master端
class Master extends Actor{
//构造代码块先被执行
println("master constructor invoked")
//定义一个map集合，用于存放worker信息
private val workerMap = new mutable.HashMap[String,WorkerInfo]()
//定义一个list集合，用于存放WorkerInfo信息，方便后期按照worker上的资源进行排序
private val workerList = new ListBuffer[WorkerInfo]
//master定时检查的时间间隔
val CHECK_OUT_TIME_INTERVAL=15000 //15秒
//prestart方法会在构造代码块执行后被调用，并且只被调用一次
override def preStart(): Unit = {
println("preStart method invoked")
//master定时检查超时的worker
//需要手动导入隐式转换
import context.dispatcher
context.system.scheduler.schedule(0 millis,CHECK_OUT_TIME_INTERVAL millis,self,CheckOutTime)
}
//receive方法会在prestart方法执行后被调用，表示不断的接受消息
override def receive: Receive = {
//master接受worker的注册信息
case RegisterMessage(workerId,memory,cores) =>{
//判断当前worker是否已经注册
if(!workerMap.contains(workerId)){
//保存信息到map集合中
val workerInfo = new WorkerInfo(workerId,memory,cores)
workerMap.put(workerId,workerInfo)
//保存workerinfo到list集合中
workerList +=workerInfo
//master反馈注册成功给worker
sender ! RegisteredMessage(s"workerId:$workerId 注册成功")
}
}
//master接受worker的心跳信息
case SendHeartBeat(workerId)=>{
//判断worker是否已经注册，master只接受已经注册过的worker的心跳信息
if(workerMap.contains(workerId)){
//获取workerinfo信息
val workerInfo: WorkerInfo = workerMap(workerId)
//获取当前系统时间
val lastTime: Long = System.currentTimeMillis()
workerInfo.lastHeartBeatTime=lastTime
}
}
case CheckOutTime=>{
//过滤出超时的worker 判断逻辑： 获取当前系统时间 - worker上一次心跳时间 >master定时检查的时间间隔
val outTimeWorkers: ListBuffer[WorkerInfo] = workerList.filter(x => System.currentTimeMillis() -x.lastHeartBeatTime > CHECK_OUT_TIME_INTERVAL)
//遍历超时的worker信息，然后移除掉超时的worker
for(workerInfo <- outTimeWorkers){
//获取workerid
val workerId: String = workerInfo.workerId
//从map集合中移除掉超时的worker信息
workerMap.remove(workerId)
//从list集合中移除掉超时的workerInfo信息
workerList -= workerInfo
println("超时的workerId:" +workerId)
}
println("活着的worker总数：" + workerList.size)
//master按照worker内存大小进行降序排列
println(workerList.sortBy(x => x.memory).reverse.toList)
}
}
}
object Master{
def main(args: Array[String]): Unit = {
//master的ip地址
val host=args(0)
//master的port端口
val port=args(1)
//准备配置文件信息
val configStr=
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider"
|akka.remote.netty.tcp.hostname = "$host"
|akka.remote.netty.tcp.port = "$port"
""".stripMargin
//配置config对象 利用ConfigFactory解析配置文件，获取配置信息
val config=ConfigFactory.parseString(configStr)
// 1、创建ActorSystem,它是整个进程中老大，它负责创建和监督actor，它是单例对象
val masterActorSystem = ActorSystem("masterActorSystem",config)
// 2、通过ActorSystem来创建master actor
val masterActor: ActorRef = masterActorSystem.actorOf(Props(new Master),"masterActor")
// 3、向master actor发送消息
//masterActor ! "connect"
}
}
7.2.2Worker类
//todo:利用akka实现简易版的spark通信框架-----Worker端
class Worker(val memory:Int,val cores:Int,val masterHost:String,val masterPort:String) extends Actor{
println("Worker constructor invoked")
//定义workerId
private val workerId: String = UUID.randomUUID().toString
//定义发送心跳的时间间隔
val SEND_HEART_HEAT_INTERVAL=10000 //10秒
//定义全局变量
var master: ActorSelection=_
//prestart方法会在构造代码块之后被调用，并且只会被调用一次
override def preStart(): Unit = {
println("preStart method invoked")
//获取master actor的引用
//ActorContext全局变量，可以通过在已经存在的actor中，寻找目标actor
//调用对应actorSelection方法，
// 方法需要一个path路径：1、通信协议、2、master的IP地址、3、master的端口 4、创建master actor老大 5、actor层级
master= context.actorSelection(s"akka.tcp://[email protected]$masterHost:$masterPort/user/masterActor")
//向master发送注册信息，将信息封装在样例类中，主要包含：workerId,memory,cores
master ! RegisterMessage(workerId,memory,cores)
}
//receive方法会在prestart方法执行后被调用，不断的接受消息
override def receive: Receive = {
//worker接受master的反馈信息
case RegisteredMessage(message) =>{
println(message)
//向master定期的发送心跳
//worker先自己给自己发送心跳
//需要手动导入隐式转换
import context.dispatcher
context.system.scheduler.schedule(0 millis,SEND_HEART_HEAT_INTERVAL millis,self,HeartBeat)
}
//worker接受心跳
case HeartBeat =>{
//这个时候才是真正向master发送心跳
master ! SendHeartBeat(workerId)
}
}
}
object Worker{
def main(args: Array[String]): Unit = {
//定义worker的IP地址
val host=args(0)
//定义worker的端口
val port=args(1)
//定义worker的内存
val memory=args(2).toInt
//定义worker的核数
val cores=args(3).toInt
//定义master的ip地址
val masterHost=args(4)
//定义master的端口
val masterPort=args(5)
//准备配置文件
val configStr=
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider"
|akka.remote.netty.tcp.hostname = "$host"
|akka.remote.netty.tcp.port = "$port"
""".stripMargin
//通过configFactory来解析配置信息
val config=ConfigFactory.parseString(configStr)
// 1、创建ActorSystem，它是整个进程中的老大，它负责创建和监督actor
val workerActorSystem = ActorSystem("workerActorSystem",config)
// 2、通过actorSystem来创建 worker actor
val workerActor: ActorRef = workerActorSystem.actorOf(Props(new
Worker(memory,cores,masterHost,masterPort)),"workerActor")
//向worker actor发送消息
workerActor ! "connect"
}
}
7.2.3WorkerInfo类
//封装worker信息
class WorkerInfo(val workerId:String,val memory:Int,val cores:Int) {
//定义一个变量用于存放worker上一次心跳时间
var lastHeartBeatTime:Long=_
override def toString: String = {
s"workerId:$workerId , memory:$memory , cores:$cores"
}
}
7.2.4样例类
trait RemoteMessage extends Serializable{
}
//worker向master发送注册信息，由于不在同一进程中，需要实现序列化
case class RegisterMessage(val workerId:String,val memory:Int,val cores:Int) extends RemoteMessage
//master反馈注册成功信息给worker,由于不在同一进程中，也需要实现序列化
case class RegisteredMessage(message:String) extends RemoteMessage
//worker向worker发送心跳 由于在同一进程中，不需要实现序列化
case object HeartBeat
//worker向master发送心跳，由于不在同一进程中，需要实现序列化
case class SendHeartBeat(val workerId:String) extends RemoteMessage
//master自己向自己发送消息，由于在同一进程中，不需要实现序列化
case object CheckOutTime