2. YARN Application 실행( CDH 를 이용하여 Local에 Spark on YARN 실행해보기 )

여러대의 Cluster 환경 구축을 위해 Vagrant 를 이용할거고, 

기본적으로 YARN을 설치하려면 hdfs 도 깔아줘야 한다. 그래서 cloudera 에서는 이러한 환경을 위해 vagrant image 를 제공해준다. 

링크 : Virtual Apache Hadoop Cluster

Vagrant 파일을 찬찬히 살펴보면 다음과 같은 내용의 설정이 되어 있다.

( 사실 그냥 실행하는데에는 이러한 사실을 몰라도 된다 ^^; )

1. ubuntu linux VM 을 4대를 띄운다.

2. host name 은 각각 vm-cluster-node1 ~ 4 이다.

3. network 는 vagrant private network 설정으로 10.211.55.100 ~ 103 으로 잡힌다.

4. vm-cluster-node1 이 하둡의 master node가 되고, 2~4가 data node 가 된다.

vagrant up 한후 http://vm-cluster-node1:7180 에 접속한다.

Cloudera Manager

admin / admin 으로 로그인을 한다.

새 호스트를 클러스터에 추가하기를 통해 4개의 VM 을 하나의 클러스터로 묶어 준다.

기본적으로 HDFS 는... 되어 있었나?? =_=; 생각이 안난다;;; ㅋㅋ 여튼 서비스 추가를 통해 본인이 필요한 서비스를 추가한다.

HDFS, YARN 을 추가한다. ( 다른거 해보고 싶으신거 있으면 해봐도 된다 :) ) 근데 Master Host의 Memory 가 아슬아슬한 모습을 볼수 있다... +_+ 많이 해볼거면 Master VM 의 Memory가 4096으로 되어 있는걸 쫌더 늘려서 쓰도록 하자.

HDFS 이용해보기 

이제 외부에서 cluster1 ( 내가 VM 으로 구성해 놓은 Hadoop Cluster ) 에 접속하기 위해 로컬에 Hadoop 을 다운로드 받는다.

링크 : http://hadoop.apache.org/#Download+Hadoop

{{hadoop_home}}/etc/hadoop 경로에 있는 다음 파일을 수정한다.

1. core-site.xml

...

	
    fs.defaultFS
    hdfs://vm-cluster-node1:8020
  

2. hdfs-site.xml

...

  
    dfs.replication
    1
  


  
    dfs.data.dir
    /data/hdfs
    true
  

  
    dfs.permissions
    false
  


  
    dfs.client.use.datanode.hostname
    true
    Whether clients should use datanode hostnames when
      connecting to datanodes.
    
  

  
    dfs.datanode.use.datanode.hostname
    true
    Whether datanodes should use datanode hostnames when
      connecting to other datanodes for data transfer.
    
  

  
    dfs.datanode.address
    vm-cluster-node1:50010
    
      The address where the datanode server will listen to.
      If the port is 0 then the server will start on a free port.
    
  

  
    dfs.datanode.http.address
    vm-cluster-node1:50075
    
      The datanode http server address and port.
      If the port is 0 then the server will start on a free port.
    
  

  
    dfs.datanode.ipc.address
    vm-cluster-node1:50020
    
      The datanode ipc server address and port.
      If the port is 0 then the server will start on a free port.
    
  

이후부터는 local 에서도 cluster1 의 hdfs 를 이용할수 있다.

$ hdfs dfs -mkdir hdfs:///sample
$ hdfs dfs -copyFormLocal hdfs:///sample ~/Documents/train.csv

위 명령어를 실행하면 권한이 없다고 나온다. cluster1 의 hdfs 가 hdfs 라는 계정에 권한이 있어 로컬에서 실행하는 경우 본인의 계정 ( 저같은 경우는 맥에서 실행하다 보기 Mac User 의 권한으로 실행이 되어 있었습니다. )

방법은 2가지가 있는데

1. root 로 실행을 하던지

2. hdfs 라는 계정을 만들어서  hdfs 계정으로 hadoop client 를 실행하던지

하면 된다.

근데... 난 귀찮으니까 일단 root 로 +_+

$ sudo hdfs dfs -mkdir hdfs:///sample
$ sudo hdfs dfs -copyFormLocal hdfs:///sample ~/Documents/train.csv
$ sudo hdfs dfs -ls hdfs:///sample

정상적으로 hdfs 에 train.csv 파일이 올라간 것을 볼수 있다. :)

replication 에 따른, 또한 Block 크기에 따른 실제 물리적인 파일 저장까지 보고 싶지만 이건 YARN 을 위한 글이니 패스하기로 해야겠다... ㅜㅠ 

Spark on YARN 실행해보기

 보통 Hadoop 책들을 보면 YARN Application....을 실행하며 Application을 RM  에 제출하고 AM 을 생성하는 이러쿵 저러쿵( 기본적인 YARN Application 의 life cycle 을 보는)한 내용을 소개해 주지만 나는 회사에서 Spark을 주로 쓰니 ( or 쓸 예정이니 ㅠㅠ ) Spark on YARN 을 실행해보기로 하자.

1. Local에 Spark 설치하기

링크 : 스파크 다운로드

위 링크에서 빌드된 스파크 파일을 다운받고 압축을 푼다. 끝.

2. ~/conf/spark-env.sh 생성 및 수정하기

 - ~/conf/spark-env.template.sh 파일을 conf 파일 아래에 복사하여 이름을 sprak-env.sh 로 바꾼다.

 - 파일 수정하기

...
# 추가
HADOOP_CONF_DIR={{your hadoop home path}}/etc/hadoop
...

3. ~/conf/spark-default.conf

 - ~/conf/sprak-default.conf.template 파일을 conf 파일 아래에 복사하여 이름을 sprak-default.conf 로 바꾼다.

 - 파일 수정하기

...
spark.master yarn
spark.deploy.mode client
spark.executor.memory 512m
...

 - spark.master : yarn cluster 를 쓰겠다는 의미.

 - spark.deploy.mode client 모드랑 cluster 모드가 있는데 client 모드는 Spark Driver 를 어디에서 실행시키는가의 기준을 놓고 다른 차이를 보여준다.

Cluster Mode

Client Mode

 - spark.executor.memory 의 dafault 값이 1g 인데... 우린 가난한(리소스가 충분하지 않은... ㅠ )  VM을 쓰니 512m 으로 최대한 낮춰서 테스트 해보자... 흙흙... ㅠㅠ

4. Pyspark Application 작성하기

 - sample.py 파일을 만든다.

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .getOrCreate()

df = spark.read.option("header","true").csv('hdfs:///sample/train.csv').cache()

df.show()

5. shell 명렁어로 실행

./spark-submit --master yarn --deploy-mode client --executor-memory 512m sample.py

 - mater, deploy-mode, executor-memory 옵션 모두 따로 주지 않아도 상관없다.

 - 왜냐면 우리는 이미 spark 를 깔면서 기본 옵션으로 위를 설정했기 때문이다.

 - 만약 deploy-mode 를 cluster 모드로 실행하고 싶은경우 위 옵션에서 --deploy-mode 를  cluster 로 바꿔주면 된다.

 - spark conf 에 먹은 옵션보다 spark-submit 쉘 명령어 옵션이 우선적으로 먹기 때문이다.

 - (속닥속닥... ) 이것도 위에서 설정해놓은 hadoop 권한 때문에 hdfs 계정으로 실행을 하던 sudo 하던지 하는 함정이 있다.... OTL 

6. Shell 로그( Spark Driver 로그), YARN 로 분석

1. Shell log 를 보면 처음 시작시에 

...
17/03/19 15:29:04 INFO Client: Application report for application_1489904829216_0001 (state: ACCEPTED)
17/03/19 15:29:04 INFO Client: 
	 client token: N/A
	 diagnostics: N/A
	 ApplicationMaster host: N/A
	 ApplicationMaster RPC port: -1
	 queue: root.users.root
	 start time: 1489904943269
	 final status: UNDEFINED
	 tracking URL: http://vm-cluster-node1:8088/proxy/application_1489904829216_0001/
	 user: root
...

이라는 log message 가 찍히는 것을 볼수 있다.

위를 통해 YARN application ID 를 알수 있고, application tracking URL 을 통해 현재 application 의 상태를 알아볼수 있다.

2. 물론 위의 tracking URL 만을 통해 얻을수 있는 정보도 많지만 쫌더 자세한 로그를 보기 위해 YARN LOGS 명령어를 이용해보도록 하자.

$ cd {{your hadoop home path}}/bin
# 위에서 알게된 application ID 를 복사한다.
$ yarn logs -applicationId application_1489904829216_0001

YARN logs 보기

1. 클라이언트는Application Master 자체를 실행하는 필요한 데이터를 포함하는 응용프로그램을 제출한다.

2. Resource Manager는 Container 할당을 책임지는 Application Master Container를 시작합니다.

Application Master Container log

17/03/18 16:49:34 INFO client.RMProxy: Connecting to ResourceManager at vm-cluster-node1/10.211.55.100:8032


Container: container_1489848543477_0004_01_000001 on vm-cluster-node2_8041
============================================================================
...

 - vm-cluster-node2 Node 에 AM Container 가 할당되고, AM container 의 ID 도 container_1489848543477_0004_01_000001 인것을 알아 낼수 있다.

3. Application Master가 Resource Manager에 등록되고 ApplicationID 를 할당받습니다. 이제부부터는 클라이언트가 Resource Manager과 통신할 수 있게 된다.

Application Master Container log

17/03/18 15:57:37 INFO ApplicationMaster: Preparing Local resources
17/03/18 15:57:38 INFO ApplicationMaster: ApplicationAttemptId: appattempt_1489848543477_0004_000001
...
17/03/18 15:57:38 INFO ApplicationMaster: Waiting for Spark driver to be reachable.
17/03/18 15:57:38 INFO ApplicationMaster: Driver now available: 10.211.55.1:56343
...

4. Application Master는 resource-request프로토콜을 통해 적절한 리소스의 Container 를 요청한다.

Application Master Container log

...
17/03/18 15:57:39 INFO YarnRMClient: Registering the ApplicationMaster
17/03/18 15:57:39 INFO YarnAllocator: Will request 2 executor container(s), each with 1 core(s) and 896 MB memory (including 384 MB of overhead)
17/03/18 15:57:39 INFO YarnAllocator: Submitted 2 unlocalized container requests.
...

5. Container 가 성공적으로 할당되면, Application Master는 Container 실행 스펙를 Node Manager에게 제공하여 Container 를 실행시킨다. 실행 스펙를 일반적으로 Container 가 Application Master 그 자체와 통신하기 위해 필요한 정보를 포함하고 있다.

6. 응용프로그램 코드는 Container 에서 실행되고 진행률, 상태 등의 정보를 응용프로그램-스펙 프로토콜을 통해 응용프로그램의 Application Master 에게 제공한다.

7. 응용프로그램 실행 중 클라이언트는 상태, 진행률 등을 얻기 위해 Application Master 와 응용프로그램-스팩 프로토콜을 통해 직접 통신한다.

Application Master Container log

...
17/03/18 15:57:39 INFO ApplicationMaster: Started progress reporter thread with (heartbeat : 3000, initial allocation : 200) intervals
17/03/18 15:57:40 INFO AMRMClientImpl: Received new token for : vm-cluster-node3:8041
17/03/18 15:57:40 INFO AMRMClientImpl: Received new token for : vm-cluster-node4:8041
17/03/18 15:57:40 INFO YarnAllocator: Launching container container_1489848543477_0004_01_000003 on host vm-cluster-node3
17/03/18 15:57:40 INFO YarnAllocator: Launching container container_1489848543477_0004_01_000004 on host vm-cluster-node4
17/03/18 15:57:40 INFO YarnAllocator: Received 2 containers from YARN, launching executors on 2 of them.
17/03/18 15:57:40 INFO ContainerManagementProtocolProxy: yarn.client.max-cached-nodemanagers-proxies : 0
17/03/18 15:57:40 INFO ContainerManagementProtocolProxy: yarn.client.max-cached-nodemanagers-proxies : 0
17/03/18 15:57:40 INFO ContainerManagementProtocolProxy: Opening proxy : vm-cluster-node3:8041
17/03/18 15:57:40 INFO ContainerManagementProtocolProxy: Opening proxy : vm-cluster-node4:8041
...

- vm-cluster-node3,4 가 YARN 어플리케이션으로 치면 AM Container 가 실행된 Node 와는 다른 Node 에서 실행된 Container들의 Host 명이다. Spark 용어로 말하자면 3,4 Host가 Executor 노드들이 된것이라고 보면 된다.

8. 일단 응용프로그램이 완료되고 모든 필요한 작업이 종료되면, Application Master 는 Resource Manager로의 등록을 해제하고 자신의 컨테이너를 다른 용도로 사용할 수 있도록 종료한다.

Application Master Container log

...
17/03/18 16:21:46 INFO ApplicationMaster$AMEndpoint: Driver terminated or disconnected! Shutting down. 10.211.55.1:56343
17/03/18 16:21:46 INFO ApplicationMaster$AMEndpoint: Driver terminated or disconnected! Shutting down. 10.211.55.1:56343
17/03/18 16:21:46 INFO ApplicationMaster: Final app status: SUCCEEDED, exitCode: 0
17/03/18 16:21:46 INFO ApplicationMaster: Unregistering ApplicationMaster with SUCCEEDED
17/03/18 16:21:46 INFO AMRMClientImpl: Waiting for application to be successfully unregistered.
17/03/18 16:21:46 INFO ApplicationMaster: Deleting staging directory hdfs://vm-cluster-node1:8020/user/root/.sparkStaging/application_1489848543477_0004
17/03/18 16:21:46 INFO ShutdownHookManager: Shutdown hook called
...

Node Manager Container log

...
17/03/18 16:21:46 INFO CoarseGrainedExecutorBackend: Driver commanded a shutdown
17/03/18 16:21:46 INFO MemoryStore: MemoryStore cleared
17/03/18 16:21:46 INFO BlockManager: BlockManager stopped
17/03/18 16:21:46 INFO ShutdownHookManager: Shutdown hook called
...

요약

로컬에 YARN 을 설치하고 YARN 위에서 Spark application 을 하나 돌려보며 job 의 처리가 분산되서 처리되는 과정을 볼 수 있었다. 이처럼 YARN 은 여러대로 묶인 Cluster 내에서 많은 Job( 또는 Application ) 이 여러대에서 잘 돌아가도록 중재해주는 것올 볼 수 있다.