Apache Spark로 Cloud Storage 커넥터 사용

이 튜토리얼은 Apache Spark에서 Cloud Storage 커넥터를 사용하는 예시 코드를 실행하는 방법을 보여줍니다.

목표

자바, Scala, Python에서 간단한 WordCount Spark 작업을 작성한 다음 Dataproc 클러스터에서 작업을 실행합니다.

비용

이 문서에서는 비용이 청구될 수 있는 다음과 같은 Google Cloud 구성요소를 사용합니다.

Compute Engine
Dataproc
Cloud Storage

프로젝트 사용량을 기준으로 예상 비용을 산출하려면 가격 계산기를 사용하세요. Google Cloud를 처음 사용하는 사용자는 무료 체험판을 사용할 수 있습니다.

시작하기 전에

아래 단계에 따라 이 튜토리얼에서 코드 실행을 준비하세요.

프로젝트를 설정합니다. 필요한 경우 로컬 머신에 Dataproc, Compute Engine 및 Cloud Storage API가 사용 설정되고 Google Cloud CLI가 설치된 상태로 프로젝트를 설정합니다.
Cloud Storage 버킷을 만듭니다. 튜토리얼 데이터를 보관하려면 Cloud Storage가 필요합니다. 사용할 준비가 되지 않은 경우 프로젝트에서 새 버킷을 만듭니다.
1. Google Cloud Console에서 Cloud Storage 버킷 페이지로 이동합니다.
  버킷 페이지로 이동
2. 버킷 만들기를 클릭합니다.
3. 버킷 만들기 페이지에서 버킷 정보를 입력합니다. 다음 단계로 이동하려면 계속을 클릭합니다.
  - 버킷 이름 지정에서 버킷 이름 지정 요구사항을 충족하는 이름을 입력합니다.
  - 데이터를 저장할 위치 선택에서 다음을 수행합니다.
    - 위치 유형 옵션을 선택합니다.
    - 위치 옵션을 선택합니다.
  - 데이터의 기본 스토리지 클래스 선택에서 스토리지 클래스를 선택합니다.
  - 객체 액세스를 제어하는 방식 선택에서 액세스 제어 옵션을 선택합니다.
  - 고급 설정(선택사항)에서 암호화 방법, 보관 정책 또는 버킷 라벨을 지정합니다.
4. 만들기를 클릭합니다.
로컬 환경 변수를 설정합니다. 로컬 머신에 환경 변수를 설정합니다. 이 튜토리얼에 사용할 Google Cloud 프로젝트 ID와 Cloud Storage 버킷의 이름을 설정합니다. 또한 기존 또는 새 Dataproc 클러스터의 이름과 리전을 입력합니다. 다음 단계에서 이 튜토리얼에서 사용할 클러스터를 만들 수 있습니다.
```
PROJECT=project-id
```
```
BUCKET_NAME=bucket-name
```
```
CLUSTER=cluster-name
```
```
REGION=cluster-region Example: "us-central1"
```
Dataproc 클러스터를 만듭니다. 아래 명령어를 실행하여 지정된 Compute Engine 영역에서 단일 노드 Dataproc 클러스터를 만듭니다.
```
gcloud dataproc clusters create ${CLUSTER} \
    --project=${PROJECT} \
    --region=${REGION} \
    --single-node
```
위의 명령어는 기본 클러스터 이미지 버전을 설치합니다. --image-version 플래그를 사용하여 클러스터의 이미지 버전을 선택할 수 있습니다. 각 이미지 버전은 특정 버전의 Spark 및 Scala 라이브러리 구성요소를 설치합니다. 자바 또는 Scala에서 Spark WordCount 작업을 준비하는 경우 작업 패키지를 준비할 때 클러스터에 설치된 Spark 및 Scala 버전을 참조합니다.
Cloud Storage 버킷에 공개 데이터를 복사합니다. 공개 데이터 Shakespeare 텍스트 스니펫을 Cloud Storage 버킷의 input 폴더에 복사합니다.
```
gcloud storage cp gs://pub/shakespeare/rose.txt \
    gs://${BUCKET_NAME}/input/rose.txt
```
자바(Apache Maven), Scala(SBT), Python 개발 환경을 설정합니다.
Cloud Shell을 사용합니다. Cloud Shell에는 Apache Maven, Python, Google Cloud CLI 등 이 튜토리얼에서 사용된 도구가 포함되어 있습니다.

Spark WordCount 작업 준비

아래의 탭을 선택하여 단계에 따라 클러스터에 제출할 작업 패키지 또는 파일을 준비합니다. 다음 작업 유형 중 하나를 준비할 수 있습니다.

자바의 Spark 작업: Apache Maven을 사용하여 JAR 패키지 빌드
Scala의 Spark 작업: SBT를 사용하여 JAR 패키지 빌드
Python의 Spark 작업(PySpark)

자바

pom.xml 파일을 로컬 머신에 복사합니다. 다음 pom.xml 파일은 Scala 및 Spark 라이브러리 종속 항목을 지정합니다. 이는 Dataproc 클러스터가 런타임 시 이러한 라이브러리를 제공함을 나타내기 위해 provided 범위를 제공합니다. 커넥터가 표준 HDFS 인터페이스를 구현하므로 pom.xml 파일은 Cloud Storage 종속 항목을 지정하지 않습니다. Spark 작업이 Cloud Storage 클러스터 파일(gs://로 시작하는 URI가 있는 파일)에 액세스할 때, 시스템이 자동으로 Cloud Storage 커넥터를 사용하여 Cloud Storage의 파일에 액세스합니다.

클러스터 이미지 버전을 확인합니다. 파일의 version 자리표시자를 대체하여 클러스터의 이미지 버전에서 사용하는 Spark 및 Scala 라이브러리 버전을 표시합니다. 참고로 spark-core_ 아티팩트 번호는 Scala major.minor 버전 번호입니다.

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>dataproc.codelab</groupId>
  <artifactId>word-count</artifactId>
  <version>1.0</version>

  <properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>Scala version, for example, 2.11.8</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_Scala major.minor.version, for example, 2.11</artifactId>
      <version>Spark version, for example, 2.3.1</version>
      <scope>provided</scope>
    </dependency>
  </dependencies>
</project>

아래 나열된 WordCount.java 코드를 로컬 머신에 복사합니다.

src/main/java/dataproc/codelab 경로를 사용하여 디렉터리 집합을 만듭니다.
```
mkdir -p src/main/java/dataproc/codelab
```
WordCount.java를 로컬 머신의 src/main/java/dataproc/codelab에 복사합니다.
```
cp WordCount.java src/main/java/dataproc/codelab
```

WordCount.java는 Cloud Storage에서 텍스트 파일을 읽고 단어 수를 계산한 후 텍스트 파일 결과를 Cloud Storage에 쓰는 자바의 간단한 Spark 작업입니다.

package dataproc.codelab;

import java.util.Arrays;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

public class WordCount {
  public static void main(String[] args) {
    if (args.length != 2) {
      throw new IllegalArgumentException("Exactly 2 arguments are required: <inputUri> <outputUri>");
    }
    String inputPath = args[0];
    String outputPath = args[1];
    JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf().setAppName("Word Count"));
    JavaRDD<String> lines = sparkContext.textFile(inputPath);
    JavaRDD<String> words = lines.flatMap(
        (String line) -> Arrays.asList(line.split(" ")).iterator()
    );
    JavaPairRDD<String, Integer> wordCounts = words.mapToPair(
        (String word) -> new Tuple2<>(word, 1)
    ).reduceByKey(
        (Integer count1, Integer count2) -> count1 + count2
    );
    wordCounts.saveAsTextFile(outputPath);
  }
}

패키지를 빌드합니다.
```
mvn clean package
```
성공적으로 빌드되면 target/word-count-1.0.jar이 생성됩니다.

패키지를 Cloud Storage로 스테이징합니다.

gcloud storage cp target/word-count-1.0.jar \
    gs://${BUCKET_NAME}/java/word-count-1.0.jar

Scala

build.sbt 파일을 로컬 머신에 복사합니다. 다음 build.sbt 파일은 Scala 및 Spark 라이브러리 종속 항목을 지정합니다. 이는 Dataproc 클러스터가 런타임 시 이러한 라이브러리를 제공함을 나타내기 위해 provided 범위를 제공합니다. 커넥터가 표준 HDFS 인터페이스를 구현하므로 build.sbt 파일은 Cloud Storage 종속 항목을 지정하지 않습니다. Spark 작업이 Cloud Storage 클러스터 파일(gs://로 시작하는 URI가 있는 파일)에 액세스할 때, 시스템이 자동으로 Cloud Storage 커넥터를 사용하여 Cloud Storage의 파일에 액세스합니다.
클러스터 이미지 버전을 확인합니다. 파일의 version 자리표시자를 대체하여 클러스터의 이미지 버전에서 사용하는 Spark 및 Scala 라이브러리 버전을 표시합니다.
```
scalaVersion := "Scala version, for example, 2.11.8"

name := "word-count"
organization := "dataproc.codelab"
version := "1.0"

libraryDependencies ++= Seq(
  "org.scala-lang" % "scala-library" % scalaVersion.value % "provided",
  "org.apache.spark" %% "spark-core" % "Spark version, for example, 2.3.1" % "provided"
)
```

word-count.scala를 로컬 머신에 복사합니다. 이는 Cloud Storage에서 텍스트 파일을 읽고 단어 수를 계산한 후 텍스트 파일 결과를 Cloud Storage에 쓰는 Java의 간단한 Spark 작업입니다.

package dataproc.codelab

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf

object WordCount {
  def main(args: Array[String]) {
    if (args.length != 2) {
      throw new IllegalArgumentException(
          "Exactly 2 arguments are required: <inputPath> <outputPath>")
    }

    val inputPath = args(0)
    val outputPath = args(1)

    val sc = new SparkContext(new SparkConf().setAppName("Word Count"))
    val lines = sc.textFile(inputPath)
    val words = lines.flatMap(line => line.split(" "))
    val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
    wordCounts.saveAsTextFile(outputPath)
  }
}

패키지를 빌드합니다.
```
sbt clean package
```
성공적으로 빌드되면 target/scala-2.11/word-count_2.11-1.0.jar이 생성됩니다.

패키지를 Cloud Storage로 스테이징합니다.

gcloud storage cp target/scala-2.11/word-count_2.11-1.0.jar \
    gs://${BUCKET_NAME}/scala/word-count_2.11-1.0.jar

Python

word-count.py를 로컬 머신에 복사합니다. 이는 PySpark를 사용하여 Cloud Storage에서 텍스트 파일을 읽고 단어 수를 계산한 후 텍스트 파일 결과를 Cloud Storage에 쓰는 Python의 간단한 Spark 작업입니다.

#!/usr/bin/env python

import pyspark
import sys

if len(sys.argv) != 3:
  raise Exception("Exactly 2 arguments are required: <inputUri> <outputUri>")

inputUri=sys.argv[1]
outputUri=sys.argv[2]

sc = pyspark.SparkContext()
lines = sc.textFile(sys.argv[1])
words = lines.flatMap(lambda line: line.split())
wordCounts = words.map(lambda word: (word, 1)).reduceByKey(lambda count1, count2: count1 + count2)
wordCounts.saveAsTextFile(sys.argv[2])

작업 제출

다음 gcloud 명령어를 실행하여 WordCount 작업을 Dataproc 클러스터에 제출합니다.

자바

gcloud dataproc jobs submit spark \
    --cluster=${CLUSTER} \
    --class=dataproc.codelab.WordCount \
    --jars=gs://${BUCKET_NAME}/java/word-count-1.0.jar \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

Scala

gcloud dataproc jobs submit spark \
    --cluster=${CLUSTER} \
    --class=dataproc.codelab.WordCount \
    --jars=gs://${BUCKET_NAME}/scala/word-count_2.11-1.0.jar \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

Python

gcloud dataproc jobs submit pyspark word-count.py \
    --cluster=${CLUSTER} \
    --region=${REGION} \
    -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/

출력 보기

작업이 완료되면 다음 gcloud CLI 명령어를 실행하여 WordCount 출력을 확인합니다.

gcloud storage cat gs://${BUCKET_NAME}/output/*

WordCount 출력은 다음과 비슷하게 표시됩니다.

(a,2)
(call,1)
(What's,1)
(sweet.,1)
(we,1)
(as,1)
(name?,1)
(any,1)
(other,1)
(rose,1)
(smell,1)
(name,1)
(would,1)
(in,1)
(which,1)
(That,1)
(By,1)

삭제

튜토리얼을 완료한 후에는 만든 리소스를 삭제하여 할당량 사용을 중지하고 요금이 청구되지 않도록 할 수 있습니다. 다음 섹션은 이러한 리소스를 삭제하거나 사용 중지하는 방법을 설명합니다.

프로젝트 삭제

비용이 청구되지 않도록 하는 가장 쉬운 방법은 튜토리얼에서 만든 프로젝트를 삭제하는 것입니다.

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

주의: 프로젝트를 삭제하면 다음과 같은 효과가 발생합니다.

프로젝트의 모든 항목이 삭제됩니다. 이 문서의 태스크에 기존 프로젝트를 사용한 경우 프로젝트를 삭제하면 프로젝트에서 수행한 다른 작업도 삭제됩니다.
커스텀 프로젝트 ID가 손실됩니다. 이 프로젝트를 만들 때 앞으로 사용할 커스텀 프로젝트 ID를 만들었을 수 있습니다. appspot.com URL과 같이 프로젝트 ID를 사용하는 URL을 보존하려면 전체 프로젝트를 삭제하는 대신 프로젝트 내에서 선택한 리소스만 삭제합니다.

여러 아키텍처, 튜토리얼, 빠른 시작을 살펴보려는 경우 프로젝트를 재사용하면 프로젝트 할당량 한도 초과를 방지할 수 있습니다.

In the Google Cloud console, go to the Manage resources page.
Go to Manage resources
In the project list, select the project that you want to delete, and then click Delete.
In the dialog, type the project ID, and then click Shut down to delete the project.

Dataproc 클러스터 삭제

프로젝트를 삭제하는 대신 프로젝트 내의 클러스터만 삭제할 수 있습니다.

Cloud Storage 버킷 삭제

Google Cloud 콘솔

In the Google Cloud console, go to the Cloud Storage Buckets page.
Go to Buckets
Click the checkbox for the bucket that you want to delete.
To delete the bucket, click Delete, and then follow the instructions.

명령줄

gcloud storage buckets delete BUCKET_NAME

다음 단계

Spark 작업 미세 조정 팁 참조하기