Spark Scala Job 제출

VPC 환경에서 이용 가능합니다.

이 가이드에서는 Spark Scala Job을 생성한 후, Cloud Hadoop 클러스터에 제출하는 방법에 대해서 설명합니다.

Scala 코드 작성 및 컴파일

Spark 애플리케이션을 Scala로 작성하고 jar로 패키징하는 방법은 다음과 같이 두 가지가 있습니다.

1. 터미널에서 Scala 사용
2. IntelliJ SBT 플러그인 사용

1. 터미널에서 Scala 사용

터미널에서 HelloScala를 출력하는 Scala 코드를 작성한 뒤 컴파일하여 jar로 패키징하는 예시를 들어 설명합니다.

Scala 바이너리 파일 다운로드

Scala 바이너리를 다운로드 하고 압축을 풀어 주십시오.

MacOS에서 homebrew를 사용하는 경우 다음과 같이 설치할 수 있습니다.

brew install scala

환경변수 설정

다음 명령어를 사용하여 실행 파일(예: .bashrc)에 SCALA_HOME 환경변수를 설정하고, PATH에 $SCALA_HOME을 추가해 주십시오.

export SCALA_HOME=/usr/local/share/scala
export PATH=$PATH:$SCALA_HOME/

Spark 애플리케이션 작성

Spark 애플리케이션을 작성하는 방법은 다음과 같습니다.

1. Scala REPL을 실행한 후 scala를 실행해 주십시오.

❯ scala
# Welcome to Scala version ...
# Type in expressions to have them evaluated.
# Type :help for more information.
# scala>

2. 다음과 같이 HelloWorld.scala class를 작성한 후 저장해 주십시오.

object HelloWorld {
  def main(args: Array[String]): Unit = {
      println("Hello, world!")
  }
}

scala> :save HelloWorld.scala
scala> :q

3. 다음 명령어를 사용하여 scalac로 컴파일해 주십시오.

❯ scalac  HelloWorld.scala

4. ls 명령어를 사용하여 정상적으로 컴파일되었는지 .class 파일을 확인해 주십시오.

❯ ls HelloWorld*.class
HelloWorld$.class HelloWorld.class

jar 파일 생성

jar 파일을 생성하는 방법은 다음과 같습니다.

1. HelloWorld*.class 파일이 있는 디렉터리로 이동한 후, 다음 명령어를 사용하여 클래스 파일을 jar로 패키징해 주십시오.

❯ jar cvfe HelloWorld.jar HelloWorld HelloWorld*.class
added manifest
adding: HelloWorld$.class(in = 670) (out= 432)(deflated 35%)
adding: HelloWorld.class(in = 645) (out= 524)(deflated 18%)

2. 패키징된 jar 파일에서 HelloWorld class가 애플리케이션의 entry point로 설정되어 있는지 MANIFEST.MF에서 확인해 주십시오.

❯ unzip -q -c HelloWorld.jar META-INF/MANIFEST.MF
Manifest-Version: 1.0
Created-By: 1.8.0_181 (Oracle Corporation)
Main-Class: HelloWorld # entry point

2. IntelliJ SBT 플러그인 사용

이 가이드에서는 Spark 애플리케이션 개발 및 디버그하기 위한 환경을 IntelliJ에 설정하고, Hello Scala라는 WordCount Job을 빌드하는 방법을 예시로 설명합니다.

• 빌드 매니저: SBT
• 예시 작성 환경: Windows OS, IntelliJ Ultimate 2022.1.4

프로젝트 생성

프로젝트를 생성하는 방법은 다음과 같습니다.

IntelliJ를 실행하여 주십시오.

1. 좌측의 Plugins 메뉴에서 Scala를 검색하여 설치해 주십시오.
   

2. 플러그인을 반영하기 위해서는 재시작이 필요합니다. [Restart IDE] 버튼을 클릭하여 IntelliJ를 재시작해 주십시오.
   

3. 홈 화면 좌측 메뉴에서 Projects를 클릭한 후, New Project를 클릭해 주십시오.
   

4. 다음과 같이 Scala와 sbt를 선택한 후, [Create] 버튼을 클릭해 주십시오.

   • 프로젝트 이름: WordCount로 지정
   • Scala와 sbt 버전을 선택
     

5. 프로젝트가 정상적으로 생성되었는지 확인해 주십시오.

   • 프로젝트가 생성되면 기본적으로 다음과 같은 구조의 디렉터리 및 파일 정보를 확인할 수 있습니다.
     • .idea: IntelliJ 구성 파일
     • project: 컴파일에 사용되는 파일
     • src : 소스코드. 애플리케이션 코드의 대부분은 src/main에 있어야 함. src/test는 테스트 스크립트를 위한 공간
     • target : 프로젝트를 컴파일하면 이 위치에 저장
     • build.sbt : SBT 구성 파일
       

SBT 라이브러리 가져오기

IntelliJ가 Spark 코드를 인식하기 위해서는 spark-core 라이브러리와 문서를 가져와야 합니다.

1. mvn repository에서 spark-core 라이브러리와 Artifact Id와 함께 호환되는 Scala 버전을 확인해 주십시오.

   

2. Target > build.sbt를 클릭한 후 스크립트 창에 아래와 같은 내용을 추가해 주십시오.

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"

3. 라이브러리를 정상적으로 가져왔는지 Build 콘솔을 통해 확인해 주십시오.
   

Group Id %% Artifact Id % Revision

Spark 애플리케이션 작성

여기서는 셰익스피어 소넷 텍스트 파일(shakespeare.txt)을 데이터 셋으로 사용하여, 소넷에 포함된 단어를 세는 WordCount 애플리케이션 작성 방법을 예시로 설명합니다.

1. shakespeare.txt을 다운로드하여 src/main/resources에 저장해 주십시오.

   • Cloud Hadoop 클러스터에서 이 애플리케이션을 실행할 때에는 S3 버킷 또는 HDFS에 데이터 셋을 업로드해서 사용합니다.
     

2. src > main를 선택해 디렉터리를 확장하고, scala 디렉터리를 마우스 오른쪽 버튼으로 클릭한 후 New > Scala Class를 클릭해 주십시오.

3. WordCount/src/main/scala 아래 새로운 클래스를 생성해 주십시오.

   • Kind: Object
     

4. 정상적으로 설정되었는지 WordCount.scala에 아래와 같은 샘플 코드를 작성한 후 실행해 주십시오.

object WordCount {
    def main(args: Array[String]): Unit = {
      println("This is WordCount application")
    }
}

5. 정상적으로 결괏값이 출력되는지 확인해 주십시오.
   

6. WordCount.scala에 적용했던 샘플 코드를 삭제한 후, 셰익스피어 소넷 텍스트 파일의 단어 수를 세는 코드를 아래와 같이 작성해 주십시오.

import org.apache.spark.{SparkConf, SparkContext}

object WordCount {

def main(args: Array[String]) : Unit = {

  //Create a SparkContext to initialize Spark
  val conf = new SparkConf()
  conf.setMaster("local")
  conf.setAppName("Word Count")
  val sc = new SparkContext(conf)

  // Load the text into a Spark RDD, which is a distributed representation of each line of text
  val textFile = sc.textFile("src/main/resources/shakespeare.txt")

  //word count
  val counts = textFile.flatMap(line => line.split(" "))
    .map(word => (word, 1))
    .reduceByKey(_ + _)

  counts.foreach(println)
  System.out.println("Total words: " + counts.count())
  counts.saveAsTextFile("/tmp/sonnetWordCount")
}

}

7. WordCount.scala를 실행하여 출력 결과를 확인해 주십시오.
   

jar 파일 생성

1. Object Storage 버킷에 데이터 셋을 업로드한 후, 소스 코드의 resource 파일 경로를 다음과 같이 변경해 주십시오.
   • 데이터 셋을 HDFS에 업로드해서 사용하려면 s3a:// 대신에 hdfs://를 사용해 주십시오.

// FROM
conf.setMaster("local")
// TO
conf.setMaster("yarn-cluster")

// FROM
val textFile = sc.textFile("src/main/resources/shakespeare.txt")
// TO
val textFile = sc.textFile("s3a://deepdrive-hue/tmp/shakespeare.txt")

// FROM
counts.saveAsTextFile("/tmp/sonnetWordCount");
// TO
counts.saveAsTextFile("s3a://deepdrive-hue/tmp/sonnetWordCount");

2. Terminal 콘솔에서 다음 명령어를 사용하여, 업데이트한 코드를 Cloud Hadoop 클러스터에 제출할 수 있도록 compiled jar로 패키징해 주십시오.
   • jar 파일에는 애플리케이션 코드와 build.sbt에 정의한 모든 dependencies가 포함되어 있습니다.
   • sbt package 명령은 $PROJECT_HOME/target/scala-2.11 아래에 wordcount_2.11-0.1.jar 파일을 생성합니다.

> cd ~/IdeaProjects/WordCount # PROJECT_HOME
> sbt package

Cloud Hadoop 클러스터에 Spark Job 제출

로컬에서 작성한 Spark 애플리케이션(.jar)을 Cloud Hadoop에 배포하여 제출하는 방법을 설명합니다.

Object Storage에 jar 업로드

Hue의 S3 브라우저 또는 Object Storage 콘솔을 사용하여 shakespeare.txt와 jar를 Object Storage 버킷에 복사해 주십시오.

• Hue 접속 및 사용에 대한 자세한 내용은 Hue 사용 가이드를 참조해 주십시오.
• Object Storage 버킷에 대한 자세한 내용은 Obeject Storage 개요 가이드를 참조해 주십시오.
  

Job 제출

jar 파일을 클러스터에 제출하는 두 가지 방법에 대해서 설명합니다.

spark.hadoop.fs.s3a.access.key=<OBEJCT-STORAGE-ACCESS-KEY>
spark.hadoop.fs.s3a.endpoint=http://kr.objectstorage.ncloud.com
spark.hadoop.fs.s3a.secret.key=<OBJECT-STORAGE-SECRET-KEY>
spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem

• Hue의 Spark Submit Jar 이용
  

• Spark 클라이언트 노드에서 제출

1. Spark 클라이언트가 설치된 클러스터 노드에서 다음과 같이 spark-submit 명령을 실행해 주십시오.

spark-submit --class WordCount --master yarn-cluster --deploy-mode cluster s3a://deepdrive-hue/tmp/wordcount_2.11-0.1.jar

2. Job 실행이 완료되면, 아래와 같이 결과가 지정한 버킷 경로 안에 저장되어 있는지 확인해 주십시오.