Livy Server 설치와 설정하기

Hadoop (+ YARN), Spark, Zeppelin의 설정을 마쳤으므로, Spark Job과 Application 제출을 위한 환경을 livy로 구성한다.

 

Bigtop을 이용한 Ambari 설치와 설정 변경은 다음 포스팅을 참고한다.

[Ambari 7] Ambari 설정과 Hadoop, Spark 설치

[Ambari 9] 데이터 탐색을 위한 하둡과 제플린 설정

[Ambari 11] Ambari 설치하면서 변경하는 설정 정리

 

 

Livy란?

Livy는 REST 인터페이스를 통해 Spark 클러스터와 쉽게 상호 작용할 수 있는 서비스이다.

• REST API를 통해 Spark에 Application이나 Code Snippet을 제출해서 Job을 생성할 수 있다.
• 제출하면서 Spark Job 실행에 필요한 Driver와 Executor에 대한 자원 할당 설정도 변경할 수 있다.
• 실행 중인 Spark Job을 중단하거나 현재 실행 중인 작업 목록을 확인하는 등 Spark 컨텍스트 관리도 가능하다.

 

Livy와 Spark의 관계에 대한 구성도는 다음과 같다. 

• REST Server가 livy이다.
• Cluster Manager는 yarn, mesos, k8s 등이 될 수 있고, Hadoop 환경에서는 대체로 yarn이다.
• Driver, Executor는 Spark을 의미한다.

 

Livy에 대한 자세한 내용은 아래 문서를 참고한다.

Apache Livy

 

 

Livy Server 설치와 설정

Livy는 Apache의 Incubator 레벨의 프로젝트로 최신 버전은 0.7.1이다.

Apache 프로젝트이기 때문에 다행히도 Apache Bigtop에서 제공하며, 버전도 0.7.1로 동일하다. 

 

Bigtop을 통한 Ambari, Livy 빌드 관련 내용은 아래 포스팅을 참고한다.

gradlew task livy 명령어를 사용하면 rpm 파일을 빌드할 수 있다.

[Ambari 5] Bigtop 설치와 Ambari 빌드

 

Ambari의 mpack에서 제공하지 않는 점은 아쉽지만, Bigtop 저장소를 통해 yum으로 설치할 수 있다.

관련 내용은 아래 포스팅을 참고한다.

[Ambari 10] Apache Bigtop 빌드 안하고 쉽게 설치하기

 

 

Ambari 기본 설정 적용

YARN (Hadoop) 환경의 Spark에 Livy를 연동할 것이므로 Ambari + Hadoop + Spark 환경이 필요하다.

아래 내용을 참고하여 환경 구성을 진행한다.

[Ambari 7] Ambari 설정과 Hadoop, Spark 설치

 

그리고 Yarn, Spark, Zeppelin 추가 설정을 진행하여 Spark이 정상 동작하는지 확인한다.

[Ambari 11] Ambari 설치하면서 변경하는 설정 정리

 

Livy 서버 설치

Ambari에서 Master 서버를 지정할 때 Spark History Server를 설치한 노드에 설치한다.

 

해당 서버에 접속한 다음 아래 명령을 실행한다.

$ ssh ambari@bdp02.bearpooh.com

$ sudo yum install livy

 

 

Livy 서버 설정

아래 내용은 다음 문서를 참고하였다.

(Optional) Installing and Configuring Livy - Hortonworks Data Platform

 

livy 계정 생성

yum으로 livy를 설치하면 livy 계정이 자동으로 생성된다.

 

아래 명령으로 livy 계정이 생성되었는지 확인한다.

$ cat /etc/passwd

# 아래와 같이 마지막 줄에 추가된다.
... 생략 ...
livy:x:997:990:Livy:/var/lib/livy:/sbin/nologin

 

hadoop 그룹 추가

아래 명령으로 hadoop 그룹에 추가한다.

$ sudo usermod -aG hadoop livy

# 추가 결과 확인
$ cat /etc/group

# 아래와 같이 hadoop 그룹의 마지막에 추가된다.
... 생략 ...

hadoop:x:1004:hive,zookeeper,spark,ams,ambari-qa,kafka,tez,hdfs,zeppelin,yarn,hcat,mapred,livy
zookeeper:x:995:
yarn:x:994:
mapred:x:993:
hive:x:992:
kafka:x:991:
livy:x:990:

 

log 폴더 소유자 변경

log 폴더의 소유자가 root:root로 되어 있기 때문에 livy:hadoop으로 변경해야 한다.

다음과 같이 진행한다.

$ sudo chown livy:hadoop /var/log/livy

$ ls -alh /var/log/livy
# 아래와 같이 livy:hadoop으로 변경된다.
drwxr-xr-x 2 livy hadoop 4.0K Apr 26 09:00 .
drwxr-xr-x. 21 root root 4.0K Apr 25 17:16 ..

 

 

livy.conf 파일 생성

livy의 설정을 관리하는 livy.conf 파일을 생성하고 아래 내용을 입력한다.

$ sudo vi /etc/livy/conf/livy.conf

livy.spark.master=yarn-cluster
livy.environment production
livy.impersonation.enabled true
livy.server.csrf_protection.enabled false
livy.server.port 8998
livy.server.session.timeout 3600000

참고사항
csrf_protection 설정이 true인 경우 Livy 서버에 POST 요청하면 400 오류를 반환한다.
해당 옵션을 false로 지정해야 한다.
https://groups.google.com/a/cloudera.org/g/hue-user/c/C_W0tOKD7cU?pli=1

 

spark-blacklist.conf 파일 생성

Spark 세션이 시작될 때 사용자가 재정의할 수 없는 속성 목록을 정의하는 spark-blacklist.conf 파일을 생성하고 아래 내용을 입력한다.

$ sudo vi /etc/livy/conf/spark-blacklist.conf

# Disallow overriding the master and the deploy mode.
spark.master
spark.submit.deployMode

# Disallow overriding the location of Spark cached jars.
spark.yarn.jar
spark.yarn.jars
spark.yarn.archive

# Don't allow users to override the RSC timeout.
livy.rsc.server.idle_timeout

 

livy-env.sh 파일 생성

Spark 경로와 Livy 실행에 필요한 환경 변수를 정의하는 livy-env.sh 파일을 생성하고 아래 내용을 입력한다.

$ sudo vi /etc/livy/conf/livy-env.sh

export SPARK_HOME=/usr/lib/spark
export LIVY_LOG_DIR=/var/log/livy
export LIVY_PID_DIR=/var/run/livy
export LIVY_SERVER_JAVA_OPTS="-Xmx2g"

 

livy의 conf 경로 권한 변경

위에 생성한 설정 파일들을 읽을 수 있도록 /etc/livy/conf 경로의 권한을 아래와 같이 744 이상으로 변경한다.

# 아래 예시는 편의상 777로 변경했다.
$ sudo chmod 777 /etc/livy/conf

$ ls -alh /etc/livy/conf
lrwxrwxrwx 1 root root 27 Apr 25 17:16 /etc/livy/conf -> /etc/alternatives/livy-conf

 

 

하둡 설정 변경

livy 서버가 Hadoop의 Yarn을 통해 Spark에 Job을 요청해야 하므로, Hadoop에 접근할 수 있도록 설정해야 한다.

Hadoop의 proxyuser 옵션에 livy의 hosts와 groups 옵션을 지정하여 Hadoop 사용자로 impersonate 한다.

 

Hadoop core-site 설정 추가

Hadoop의 core-site.xml 파일의 설정을 변경해야 하며, Ambari의 HDFS - CONFIGS 탭에서 진행한다.

 

아래와 같이 HDFS - CONFIGS 탭으로 이동한다.

 

스크롤을 내려서 Custom hdfs-site를 클릭하여 메뉴를 확장한다.

 

하단의 Add Property를 클릭하면 다음 화면이 나타난다.

 

 

아래와 같이 proxyuser 옵션에 livy의 groups와 hosts 옵션을 추가한다. 

한번에 하나씩 두번 입력해야 한다.

구분	Groups	Hosts
Key	hadoop.proxyuser.livy.groups	hadoop.proxyuser.livy.hosts
Value	*	*
Property Type	TEXT	TEXT

 

입력이 완료되면 다음과 같이 추가된다.

 

영향 받는 서비스 재시작

하단의 SAVE 버튼을 클릭하면 좌측 서비스 목록에 재시작이 필요한 항목이 표시된다.

 

각 서비스들을 선택하고 우측의 RESTART 버튼을 펼쳐서 Restart All Affected 버튼을 클릭한다.

일반적으로 HDFS, YARN, MapReduce2, Hive, Spark 서비스를 재시작해야 한다.

 

 

Livy 서버 시작과 연결 확인

설치와 설정이 완료되면 아래와 같이 Livy 서버를 시작한다.

$ ssh ambari@bdp02.bearpooh.com

$ sudo /usr/lib/livy/bin/livy-server start
$ sudo /usr/lib/livy/bin/livy-server status
livy-server is running (pid: 32122)

 

livy.conf 파일에서 지정한 8998 포트로 접속하면 아래와 같이 확인된다.

 

Spark Job 제출 테스트

Livy 서버와 Spark의 연결이 잘 되었는지 확인하기 위해 예제코드를 실행한다.

 

예제 코드 준비

예제코드는 아래 사이트를 참고하였다.

Cloud Hadoop의 Apache Livy 사용하여 Spark Job 제출하기

 

전체적인 기능 구성은 다음과 같다.

• Spark 세션 생성
• Spark 세션 상태 확인
• 테스트 코드 1 실행 (1 + 1 결과 출력)
• 테스트 코드 2 실행 (원주율 계산)
• Spark 세션 종료

 

코드 전문은 다음을 참고한다. (펼치기)

$ python livy.py http://bdp02.bearpooh.com:8998

#-*- coding:utf-8 -*-

import json, pprint, requests, textwrap, time, sys

# Livy2 접속 정보 입력
if len(sys.argv) < 2:
    print('ERROR : Livy 서버 접속 정보를 입력해 주세요')
    print(' - Usage: python {0} http://호스트명:포트'.format(sys.argv[0]))
    sys.exit(1)
host = sys.argv[1]

# 헤더 정보
headers = {'Content-Type': 'application/json'}

# Spark 세션 생성
data = {'kind': 'spark'}
r = requests.post(host + '/sessions', data=json.dumps(data), headers=headers)
print("Created " + r.headers['location'])

# Spark 세션 상태 확인
state = "notIdle"
session_url = host + r.headers['location']
sys.stdout.write('Waiting for session state to idle')
while state != 'idle':
    r = requests.get(session_url, headers=headers)
    state = r.json()['state']
    sys.stdout.write('.')
    sys.stdout.flush()
    time.sleep(1)
sys.stdout.write('\rSession State is Ready!!!!!!!!!!!!!!\n')
sys.stdout.flush()

# 테스트 코드 1
statements_url = session_url + '/statements'
data = {'code': '1 + 1'}
r = requests.post(statements_url, data=json.dumps(data), headers=headers)
statement_url = host + r.headers['location']
print('=' * 80)
print(statement_url)
print('Request: {0}'.format(data['code']))

output = None
while output == None:
    r = requests.get(statement_url, headers=headers)
    ret = r.json()
    if ret['output'] == None:
        time.sleep(1)
        continue
    if 'output' in ret and 'data' in ret['output']:
        output = ret['output']['data']['text/plain']

print('-' * 80)
print(output)

# 테스트 코드 2
data = {
    'code': textwrap.dedent("""
        val NUM_SAMPLES = 100000;
        val count = sc.parallelize(1 to NUM_SAMPLES).map { i =>
            val x = Math.random();
            val y = Math.random();
            if (x*x + y*y < 1) 1 else 0
        }.reduce(_ + _);
        println(\"Pi is roughly \" + 4.0 * count / NUM_SAMPLES)
    """)
}

r = requests.post(statements_url, data=json.dumps(data), headers=headers)
statement_url = host + r.headers['location']
print('=' * 80)
print(statement_url)
print('Request: {0}'.format(data['code']))

output = None
while output == None:
    r = requests.get(statement_url, headers=headers)
    ret = r.json()
    if ret['output'] == None:
        time.sleep(1)
        continue
    if 'output' in ret and 'data' in ret['output']:
        output = ret['output']['data']['text/plain']

print('-' * 80)
print(output)

# Spark 세션 종료
print('=' * 80)
r = requests.delete(session_url, headers=headers)
print('{0} {1}'.format(r.json()['msg'], session_url))​

 

 

예제 코드 실행과 결과 확인

위의 코드를 실행하면 아래와 같이 결과가 출력된다.

$ python livy.py http://bdp02.bearpooh.com:8998
Created /sessions/3
Waiting for session state to idle......................
.........................Session State is Ready!!!!!!!!!!!!!!
================================================================================
http://bdp02.bearpooh.com:8998/sessions/3/statements/0
Request: 1 + 1
--------------------------------------------------------------------------------
res0: Int = 2

================================================================================
http://bdp02.bearpooh.com:8998/sessions/3/statements/1
Request:
val NUM_SAMPLES = 100000;
val count = sc.parallelize(1 to NUM_SAMPLES).map { i =>
val x = Math.random();
val y = Math.random();
if (x*x + y*y < 1) 1 else 0
}.reduce(_ + _);
println("Pi is roughly " + 4.0 * count / NUM_SAMPLES)

--------------------------------------------------------------------------------
NUM_SAMPLES: Int = 100000
count: Int = 78523
Pi is roughly 3.14092

================================================================================

+deleted http://bdp02.bearpooh.com:8998/sessions/3

 

Spark History 서버에서 아래와 같이 실행 이력을 확인할 수 있다.

Spark History Server는 Ambari를 설치할 때 Slave Node (Client)를 구성할 때 결정한다.

아래 포스팅의 Assign Slaves and Clients 부분을 참고한다.

[Ambari 7] Ambari 설정과 Hadoop, Spark 설치