간단한 Spark Application 만들어 보기

기본적인 Scala 개발 환경 구성이 완료되었으면 간단한 Spark 어플리케이션을 만들어본다.

해당 과정을 통해 sbt로 라이브러리를 잘 받아오는지, Spark 라이브러리가 잘 동작하는지 확인할 수 있다.

 

아래 내용들은 리눅스 운영체제를 기반으로 설명한다.

윈도우 환경인 경우 WSL을 통해 진행한다.

 

IntelliJ 설치와 초기 설정은 아래 포스팅을 참고한다.

Ubuntu 22.04에 IntelliJ 설치하기

윈도우 WSL 환경에서 IntelliJ 사용하기

 

 

프로젝트 생성

Scala 프로젝트 생성하기

초기 화면에서 New Project를 클릭한다.

 

생성할 프로젝트의 정보를 입력한다.

• Name : SparkExample을 입력한다.
• JDK : 1.8 버전을 선택한다.
• sbt : 1.9.0 버전을 선택한다. Download sources 항목은 체크 해제한다.
• Scala : 2.12.18 버전을 선택한다. Download sources 항목에 체크한다.

 

 

생성 된 파일 확인

IntelliJ가 생성하는 기본 파일들을 확인한다.

 

build.sbt

프로젝트의 기본 옵션을 결정하는 build.sbt 파일을 확인한다.

version에는 0.1.0-SNAPSHOT이 기본 설정된다.

scalaVersion은 2.12.18 버전이 지정 된 것을 확인할 수 있다.

 

build.properties

sbt의 버전을 지정하는 build.properties 파일을 확인한다.

sbt.version 항목에 1.9.0 버전이 지정된 것을 확인할 수 있다.

 

프로젝트 설정 변경

IntelliJ의 프로젝트 설정 변경 방법은 아래 포스팅을 참고한다.

IntelliJ 초기 환경 설정

 

아래 항목들을 중심으로 진행한다.

• Key Map (키맵) 변경
• sbt 저장소 설정 변경
• sbt의 JVM 인증서 설정 변경
• sbt 기능 사용 범위 설정

 

 

코드 작성하기

프로젝트 생성과 기본 설정이 완료되었으므로 Spark 어플리케이션 예제코드를 작성한다.

 

예제 코드와 JSON 데이터는 아래 블로그의 내용을 참고했다.

Spark 개발 환경 구축하기 - Scala, IntelliJ, SBT

 

build.sbt 수정

먼저 build.sbt 파일을 수정한다.

파이썬의 setup.py와 같이 어플리케이션의 기본 정보와 사용하는 라이브러리 정보들을 입력한다.

Spark은 3.2.3 버전으로 지정하여, spark-core, spark-sql 라이브러리를 3.2.3 버전으로 다운로드한다.

 

코드는 다음과 같다.

import Keys._

val sparkVersion = "3.2.3"

lazy val root = (project in file("."))
  .settings(
    name := "sparkexample",
    version := "0.1.0.1-SNAPSHOT",    // 개발중인 버전이므로 SNAPSHOT 추가
    versionScheme := Some("semver-spec"),    // Semantic 버전 체계 사용
    organization := "com.bearpooh.test",    // 소속 조직 정보 입력 (배포시 저장 폴더)
    scalaVersion := "2.12.18",    // 사용하는 Scala 버전
    libraryDependencies ++= Seq(    // 라이브러리 의존성 설정 (Spark 관련 라이브러리)
      "org.apache.spark" %% "spark-core" % sparkVersion % "provided",
      "org.apache.spark" %% "spark-sql" % sparkVersion % "provided"
    ),
    publishMavenStyle := true    // Java의 Maven 형식으로 배포
  )

 

 

json 파일 생성

src의 main 폴더에 resources 폴더를 생성한다.

main 폴더 우클릭 > New > Directory를 선택한다.

 

생성된 resources 폴더를 우클릭하고 Mark Directory as > Resources Root를 클릭한다.

일반 폴더 아이콘의 모양이 바뀐 것을 확인할 수 있다.

 

다시 resources 폴더를 우클릭하고 New > File을 선택한다.

 

파일명에 log.json을 입력하면 빈 파일이 생성된다.

해당 파일에 아래 내용을 작성한다.

{"id":"2688425498","type":"buy","created_at":"2013-08-13T22:58:08Z"}
{"id":"2688425496","type":"view","created_at":"2013-08-13T22:59:08Z"}
{"id":"2688425492","type":"view","created_at":"2013-08-13T23:01:08Z"}
{"id":"2688425491","type":"buy","created_at":"2013-08-13T22:02:08Z"}
{"id":"2688425493","type":"view","created_at":"2013-08-13T22:03:08Z"}
{"id":"2688425494","type":"view","created_at":"2013-08-13T22:06:08Z"}
{"id":"2688425495","type":"view","created_at":"2013-08-13T22:08:08Z"}

 

아래와 같이 작성하면 된다.

 

 

main 코드 작성

이제 main 함수의 역할을 담당할 SparkExampleApp.scala 파일을 작성한다.

 

Package 경로 생성

main 하위의 scala 폴더를 우클릭하고 Package를 클릭한다.

 

사용할 Package 경로를 입력한다. 각자 속한 조직의 FQDN 또는 보유한 도메인을 활용한다.

Java 계열에서는 패키지 경로를 FQDN을 역순으로 작성하는 관례를 따른다.

 

개인적으로 bearpooh.com 도메인을 보유하고 있으므로 com.bearpooh로 시작하도록 작성했다.

또한 test로 작성한 Spark 예제코드이기 때문에 test.sparkexample을 추가했다.

최종적으로 패키지 경로는 com.bearpooh.test.sparkexample로 입력했다.

 

패키지 경로를 생성하면 . (dot)을 기준으로 계층 구조로 폴더들이 생성된다.

IntelliJ에서는 단일 폴더로 다룰수 있도록 묶어서 표시해준다.

 

참고로 IntelliJ에서 이러한 패키지 폴더를 편하게 관리할 수 있도록  Package Prefix 기능을 제공한다.

아직 제대로 사용해보지 않았으므로, 향후 사용해보고 다시 정리한다.

 

SparkExampleApp.scala 파일 생성

SparkExampleApp.scala 파일을 생성한다.

 

Scala에서 파일명과 클래스명은 각 단어들의 첫글자들을 모두 대문자로 표기하는 파스칼 (Pascal) 케이스를 사용한다.

함수명과 변수명은 파스칼 케이스에서 첫 단어의 첫 글자를 소문자로 표기하는 카멜 (Camel) 케이스를 사용한다.

 

생성한 패키지 폴더를 우클릭하고 New > Scala Class를 선택한다.

 

Object를 선택하고 SparkExampleApp을 입력한다.

 

아래와 같이 SparkExampleApp.scala 파일이 생성된다.

 

 

SparkExampleApp.scala의 main 함수 작성

아래 코드를 참고하여 main 함수를 작성한다.

설명은 주석을 참고한다.

package com.bearpooh.test.sparkexample

import org.apache.log4j.LogManager
import org.apache.spark.sql.SparkSession

// main 함수가 실행되는 경우 object로 생성
object SparkExampleApp {
  // args는 실행될 때 전달 되는 인자 값 (Arguments)
  def main(args: Array[String]): Unit = {
    // Spark Session 생성
    val spark = SparkSession.builder()
      .appName("SparkExample")
      .master("local[*]")
      .getOrCreate()

    // 결과를 출력하기 위한 Logger 생성
    // 출력문인 println도 가능하지만 가급적 logger를 사용하는 것을 권장한다.
    val logger = LogManager.getLogger(this.getClass.getName)

    val logPath = "src/main/resources/log.json"

    // json 파일을 DataFrame으로 읽기
    val logDF = spark.read.json(logPath)
    // 전체 레코드 개수 세기
    val logCount = logDF.count()

    // 전체 레코드 개수 출력
    logger.info("log count: " + logCount)
  }
}

 

실제 작성 결과는 다음과 같다.

 

 

참고자료

위의 내용은 아래 문서 내용을 참고하여 작성하였다.

Spark 개발 환경 구축하기 - Scala, IntelliJ, SBT

How to do effective logging in Spark application