Spark 是一个 通用的分布式计算系统,既然是分布式的,必然存在很多节点之间的通信,那么 Spark 不同组件之间就会通过 RPC(Remote Procedure Call)进行点对点通信。
Spark 的 RPC 主要在两个模块中:
1,spark-core 中,主要承载了更好的封装 server 和 client 的作用,以及和 scala 语言的融合,它依赖 spark-network-common 模块;
2,spark-network-common 中,该模块是 java 语言写的,最新版本是基于 Netty 开发的;
Spark 早期版本中使用 Netty 通信框架做大块数据的传输,使用 Akka 用作 RPC 通信。自 Spark2.0 之后已经把 Akka 框架玻璃出去了(详见SPARK-5293),是因为很多用户会使用 Akka 做消息传递,会与 Spark 内嵌的版本产生冲突。在 Spark2.0 之后,基于底层的 spark-network-commen 模块实现了一个类似 Akka Actor 消息传递模式的 scala 模块,封装在 spark-core 中。
看一张 UML 图,图内展示了 Spark RPC 模块内的类的关系,白色的是 spark-core 中的类,黄色的 spark-common 中的类:
整个 Spark 的 RPC 模块大概有几个主要的类构成:
1,RpcEndPonit 和 RpcCallContext,RpcEndPoint 是一个可以相应请求的服务,类似于 Akka 中的 Actor。其中有 receive 方法用来接收客户端发送过来的信息,也有 receiveAndReply 方法用来接收并应答,应答通过 RpcContext 回调。可以看下面代码:
def receive: PartialFunction[Any, Unit] = {
case _ => throw new RpcException(self + " does not implement 'receive'")
}
def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
case _ => context.sendFailure(new RpcException(self + " won't reply anything"))
}
2,RpcEndpointRef,类似于 Akka 中的 ActorRef,是 RpcEndPoint 的引用,持有远程 RpcEndPoint 的地址名称等,提供了 send 方法和 ask 方法用于发送请求。可以看看 RpcEndPoint 内部的成员变量和方法:
/**
* return the address for the [[RpcEndpointRef]]
*/
def address: RpcAddress
def name: String
/**
* Sends a one-way asynchronous message. Fire-and-forget semantics.
*/
def send(message: Any): Unit
/**
* Send a message to the corresponding [[RpcEndpoint.receiveAndReply)]] and return a [[Future]] to
* receive the reply within the specified timeout.
*
* This method only sends the message once and never retries.
*/
def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T]
3,RpcEnv 和 NettyRpcEnv
RpcEnv 类似于 ActorSystem,服务端和客户端都可以使用它来做通信。
对于 server 端来说,RpcEnv 是 RpcEndpoint 的运行环境,负责 RpcEndPoint 的生命周期管理,解析 Tcp 层的数据包以及反序列化数据封装成 RpcMessage,然后根据路由传送到对应的 Endpoint;
对于 client 端来说,可以通过 RpcEnv 获取 RpcEndpoint 的引用,也就是 RpcEndpointRef,然后通过 RpcEndpointRef 与对应的 Endpoint 通信。
RpcEnv 中有两个最常用的方法:
// 注册endpoint,必须指定名称,客户端路由就靠这个名称来找endpoint
def setupEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef
// 拿到一个endpoint的引用
def setupEndpointRef(address: RpcAddress, endpointName: String): RpcEndpointRef
NettyRpcEnv 是 spark-core 和 spark-network-common 的桥梁,内部 leverage 底层提供的通信能力,同事包装了一个类 Actor 的语义。
4,Dispatcher ,NettyRpcEnv 中包含 Dispatcher,主要针对服务端,帮助路由到指定的 RpcEndPoint,并调用起业务逻辑。
参考: