Spark 中的 RPC

1,099 阅读3分钟

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,并调用起业务逻辑。

参考:

1,blog.csdn.net/justlpf/art…

2,zhuanlan.zhihu.com/p/28893155