[Spark] RegisterApplication RPC 알아보기

- 목차

 

RegisterApplication RPC 란 ?

RegisterApplication 이란 Driver 가 Master Node 나 Resource Manager 에게 어플리케이션을 등록하기 위한 RPC 요청입니다.

Spark Driver가 실행되면, Spark Master 와 통신하여 "새로운 Application을 등록하고 싶다" 라는 메시지를 보냅니다.

즉, Driver 가 곧 Application 과 일대일로 매칭되는 단위라고 생각하셔도 됩니다.

 

 

RegisterApplication RPC 요청은 아래의 두가지 데이터를 Master 에게 전달합니다.

ApplicationDescription 은 Executor 의 스펙을 정의합니다.

몇 개의 Executor 를 사용하며, 각 Executor 의 CPU 와 Memory 사용량을 제한할 수 있습니다.

또한 RpcEndpointRef 정보를 제공하여 Driver 의 네트워크 주소 정보를 함께 전달합니다.

private[spark] case class ApplicationDescription(
    name: String,
    maxCores: Option[Int],
    command: Command,
    appUiUrl: String,
    defaultProfile: ResourceProfile,
    eventLogDir: Option[URI] = None,
    eventLogCodec: Option[String] = None,
    initialExecutorLimit: Option[Int] = None,
    user: String = System.getProperty("user.name", "<unknown>")) {

  def memoryPerExecutorMB: Int = defaultProfile.getExecutorMemory.map(_.toInt).getOrElse(1024)
  def coresPerExecutor: Option[Int] = defaultProfile.getExecutorCores
  def resourceReqsPerExecutor: Seq[ResourceRequirement] =
    ResourceUtils.executorResourceRequestToRequirement(
      defaultProfile.getCustomExecutorResources().values.toSeq.sortBy(_.resourceName))
}

private[spark] abstract class RpcEndpointRef(conf: SparkConf)
  extends Serializable with Logging {
  private[this] val defaultAskTimeout = RpcUtils.askRpcTimeout(conf)
  def address: RpcAddress
  def name: String
}

 

Driver 는 Spark Worker Node 의 분산 환경에서 실행할 Executor 들이 생성되도록 요청할 수 있고,

생성된 Executor 는 Driver 와 통신을 수행하게 됩니다.

따라서 Executor 의 스펙과 Driver 의 주소 정보를 함께 전달합니다.

 

아래는 Driver 에서 Master Node 로 전달되는 TCP Packet 의 내용입니다.

192.168.45.2 -> 127.0.0.1 로 RegisterApplication 요청이 전달됩니다.

제 로컬 환경의 Private IP 가 192.168.45.2 이고, 127.0.0.1:7077 인 Master 에게 RegisterApplication 요청을 전달합니다.

 

22:06:10.152475 IP 172.23.0.1.56318 > master.7077: Flags [P.], seq 196:2784, ack 69, win 512, options [nop,nop,TS val 1176454428 ecr 3126141700], length 2588
	0x0000:  4500 0a50 f82c 4000 4006 e049 ac17 0001  E..P.,@.@..I....
	0x0010:  ac17 0002 dbfe 1ba5 7092 0ac9 af9e 21fc  ........p.....!.
	0x0020:  8018 0200 6274 0000 0101 080a 461f 451c  ....bt......F.E.
	0x0030:  ba55 2304 0000 0000 0000 0a1c 0900 000a  .U#.............
	0x0040:  0f01 000e 3139 322e 3136 382e 3435 2e32  ....192.168.45.2
	0x0050:  3038 0000 e567 0100 0931 3237 2e30 2e30  08...g...127.0.0
	0x0060:  2e31 0000 1ba5 0006 4d61 7374 6572 aced  .1......Master..
	0x0070:  0005 7372 003a 6f72 672e 6170 6163 6865  ..sr.:org.apache
	0x0080:  2e73 7061 726b 2e64 6570 6c6f 792e 4465  .spark.deploy.De
	0x0090:  706c 6f79 4d65 7373 6167 6573 2452 6567  ployMessages$Reg
	0x00a0:  6973 7465 7241 7070 6c69 6361 7469 6f6e  isterApplication
	0x00b0:  b3bd 8dd3 0609 1fef 0200 024c 000e 6170  ...........L..ap
    // .. 중략

 

등록된 Application 들은 아래와 같이 Spark Master 의 UI 에서 확인이 가능합니다.

 

 

RegisterApplication 실행 과정.

Spark의 RegisterApplication RPC는 아래와 같은 단계를 거쳐 실행됩니다.

 

1. Driver에서 RegisterApplication 메시지 생성 및 전송

Spark 애플리케이션이 실행되면, Driver 는 RegisterApplication 메시지를 생성하여 Master 에게 전송합니다.

아래의 이미지처럼 Driver 는 Master 에게 RegisterApplication 을 요청하며,

Driver 의 자신과 통신할 수 있도록 IP/Port 를 넘기며, Executor Spec 을 함께 전송합니다.

 

출처 : https://ymgd.github.io/codereader/2018/03/24/Spark2.X-Deploy%E6%A8%A1%E5%9D%97%E8%A7%A3%E6%9E%90/

 

 

2. Master 는 Worker Node 에게 Executor 생성 명령

RegisterApplication 요청을 전달받은 Master 는 Worker 들에게 LaunchExecutor RPC 를 전달합니다.

Executor 의 갯수, CPU/Memory 사용량 등을 전달하여 Worker 로부터 Executor 실행하도록 합니다.

 

아래는 Master 가 Worker 에게 LaunchExecutor 명령를 전달하는 TCP Packet 의 정보입니다.

 

22:06:10.171385 IP master.7077 > worker1.spark.45434: Flags [P.], seq 7057:8844, ack 20932, win 658, options [nop,nop,TS val 1487084964 ecr 3724864030], length 1787
	0x0000:  4500 072f 7dd8 4000 4006 5dbd ac17 0002  E../}.@.@.].....
	0x0010:  ac17 0003 1ba5 b17a e06a 110f e7bc 848c  .......z.j......
	0x0020:  8018 0292 5f55 0000 0101 080a 58a3 1da4  ...._U......X...
	0x0030:  de04 ea1e 0100 066d 6173 7465 7200 001b  .......master...
	0x0040:  a501 000a 3137 322e 3233 2e30 2e33 0000  ....172.23.0.3..
	0x0050:  983d 0006 576f 726b 6572 aced 0005 7372  .=..Worker....sr
	0x0060:  0035 6f72 672e 6170 6163 6865 2e73 7061  .5org.apache.spa
	0x0070:  726b 2e64 6570 6c6f 792e 4465 706c 6f79  rk.deploy.Deploy
	0x0080:  4d65 7373 6167 6573 244c 6175 6e63 6845  Messages$LaunchE
	0x0090:  7865 6375 746f 7277 08ee 4d5f f9b4 8502  xecutorw..M_....
	0x00a0:  0007 4900 0563 6f72 6573 4900 0665 7865  ..I..coresI..exe
	0x00b0:  6349 6449 0006 6d65 6d6f 7279 4c00 0761  cIdI..memoryL..a
	0x00c0:  7070 4465 7363 7400 304c 6f72 672f 6170  ppDesct.0Lorg/ap

 

 

3. Driver 와 Executor 의 통신

마지막으로 Driver 는 Executor 에게 실행해야할 Job 과 Task 등을 전달합니다.

Executor 는 전달받은 Task 들을 실행하고 Driver 에게 결과를 응답합니다.

Driver 는 RegisterApplication 단계에서 Driver 의 주소 정보를 전달하였는데요.

Master 로부터 Driver 의 정보를 전달받은 Executor 는 Driver 와 통신을 수행할 수 있게 됩니다.

이 단계에서 LaunchTask 요청이 Driver 로부터 Worker 의 Executor 에게 전달합니다.

 

22:35:11.018454 IP 192.168.45.208.59128 > bcfcbbd51f99.42146: Flags [.], seq 16275:17723, ack 14611, win 4096, options [nop,nop,TS val 2917605211 ecr 206986842], length 1448
E.......?.E...-.........n\,."8......U......
...[.V^Z........	.......192.168.45.208.......
Executor....sr.Jorg.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages$LaunchTask@j>
// ... 중략