Apache Spark로 Cloud Storage 커넥터 사용

목표

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

비용

이 문서에서는 비용이 청구될 수 있는 Google Cloud구성요소( )를 사용합니다.

• Compute Engine
• Dataproc
• Cloud Storage

프로젝트 사용량을 기준으로 예상 비용을 산출하려면 가격 계산기를 사용합니다.

시작하기 전에

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

1. 프로젝트를 설정합니다. 필요한 경우 로컬 머신에 Dataproc, Compute Engine 및 Cloud Storage API가 사용 설정되고 Google Cloud CLI가 설치된 상태로 프로젝트를 설정합니다.

   Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

   In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Go to project selector

   Verify that billing is enabled for your Google Cloud project.

   Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

   Enable the APIs

   Create a service account:

   1. In the Google Cloud console, go to the Create service account page.

      Go to Create service account
   2. Select your project.

   3. In the Service account name field, enter a name. The Google Cloud console fills in the Service account ID field based on this name.

      In the Service account description field, enter a description. For example, Service account for quickstart.

   4. Click Create and continue.

   5. Grant the Project > Owner role to the service account.

      To grant the role, find the Select a role list, then select Project > Owner.

   6. Click Continue.

   7. Click Done to finish creating the service account.

      Do not close your browser window. You will use it in the next step.

   Create a service account key:

   1. In the Google Cloud console, click the email address for the service account that you created.
   2. Click Keys.
   3. Click Add key, and then click Create new key.
   4. Click Create. A JSON key file is downloaded to your computer.
   5. Click Close.

   Set the environment variable GOOGLE_APPLICATION_CREDENTIALS to the path of the JSON file that contains your credentials. This variable applies only to your current shell session, so if you open a new session, set the variable again.

   Example: Linux or macOS

   export GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   For example:

   export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

   Example: Windows

   For PowerShell:

   $env:GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   For example:

   $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\service-account-file.json"

   For command prompt:

   set GOOGLE_APPLICATION_CREDENTIALS=KEY_PATH

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   Install the Google Cloud CLI.

   외부 ID 공급업체(IdP)를 사용하는 경우 먼저 제휴 ID로 gcloud CLI에 로그인해야 합니다.

   gcloud CLI를 초기화하려면 다음 명령어를 실행합니다.

   gcloud init

   In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Go to project selector

   Verify that billing is enabled for your Google Cloud project.

   Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

   Enable the APIs

   Create a service account:

   1. In the Google Cloud console, go to the Create service account page.

      Go to Create service account
   2. Select your project.

   3. In the Service account name field, enter a name. The Google Cloud console fills in the Service account ID field based on this name.

      In the Service account description field, enter a description. For example, Service account for quickstart.

   4. Click Create and continue.

   5. Grant the Project > Owner role to the service account.

      To grant the role, find the Select a role list, then select Project > Owner.

   6. Click Continue.

   7. Click Done to finish creating the service account.

      Do not close your browser window. You will use it in the next step.

   Create a service account key:

   1. In the Google Cloud console, click the email address for the service account that you created.
   2. Click Keys.
   3. Click Add key, and then click Create new key.
   4. Click Create. A JSON key file is downloaded to your computer.
   5. Click Close.

   Set the environment variable GOOGLE_APPLICATION_CREDENTIALS to the path of the JSON file that contains your credentials. This variable applies only to your current shell session, so if you open a new session, set the variable again.

   Example: Linux or macOS

   export GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   For example:

   export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

   Example: Windows

   For PowerShell:

   $env:GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   For example:

   $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\service-account-file.json"

   For command prompt:

   set GOOGLE_APPLICATION_CREDENTIALS=KEY_PATH

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   Install the Google Cloud CLI.

   외부 ID 공급업체(IdP)를 사용하는 경우 먼저 제휴 ID로 gcloud CLI에 로그인해야 합니다.

   gcloud CLI를 초기화하려면 다음 명령어를 실행합니다.

   gcloud init

   1. Cloud Storage 버킷을 만듭니다. 튜토리얼 데이터를 보관하려면 Cloud Storage가 필요합니다. 사용할 준비가 되지 않은 경우 프로젝트에서 새 버킷을 만듭니다.

      1. In the Google Cloud console, go to the Cloud Storage Buckets page.

         Go to Buckets

      2. Click add_box Create.
      3. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
         1. In the Get started section, do the following:
            • Enter a globally unique name that meets the bucket naming requirements.
            • To add a bucket label, expand the Labels section (expand_more), click add_box Add label, and specify a key and a value for your label.
         2. In the Choose where to store your data section, do the following:
            1. Select a Location type.
            2. Choose a location where your bucket's data is permanently stored from the Location type drop-down menu.
               • If you select the dual-region location type, you can also choose to enable turbo replication by using the relevant checkbox.
            3. To set up cross-bucket replication, select Add cross-bucket replication via Storage Transfer Service and follow these steps:

               Set up cross-bucket replication

               1. In the Bucket menu, select a bucket.

               2. In the Replication settings section, click Configure to configure settings for the replication job.

                  The Configure cross-bucket replication pane appears.

                  • To filter objects to replicate by object name prefix, enter a prefix that you want to include or exclude objects from, then click add Add a prefix.
                  • To set a storage class for the replicated objects, select a storage class from the Storage class menu. If you skip this step, the replicated objects will use the destination bucket's storage class by default.
                  • Click Done.
         3. In the Choose how to store your data section, do the following:
            1. Select a default storage class for the bucket or Autoclass for automatic storage class management of your bucket's data.
            2. To enable hierarchical namespace, in the Optimize storage for data-intensive workloads section, select Enable hierarchical namespace on this bucket.
         4. In the Choose how to control access to objects section, select whether or not your bucket enforces public access prevention, and select an access control method for your bucket's objects.
         5. In the Choose how to protect object data section, do the following:
            • Select any of the options under Data protection that you want to set for your bucket.
              • To enable soft delete, click the Soft delete policy (For data recovery) checkbox, and specify the number of days you want to retain objects after deletion.
              • To set Object Versioning, click the Object versioning (For version control) checkbox, and specify the maximum number of versions per object and the number of days after which the noncurrent versions expire.
              • To enable the retention policy on objects and buckets, click the Retention (For compliance) checkbox, and then do the following:
                • To enable Object Retention Lock, click the Enable object retention checkbox.
                • To enable Bucket Lock, click the Set bucket retention policy checkbox, and choose a unit of time and a length of time for your retention period.
            • To choose how your object data will be encrypted, expand the Data encryption section (expand_more), and select a Data encryption method.
      4. Click Create.

   2. 로컬 환경 변수를 설정합니다. 로컬 머신에 환경 변수를 설정합니다. 이 튜토리얼에 사용할 Cloud Storage 버킷의 이름과 Google Cloud 프로젝트 ID를 설정합니다. 또한 기존 또는 새 Dataproc 클러스터의 이름과 리전을 입력합니다. 다음 단계에서 이 튜토리얼에서 사용할 클러스터를 만들 수 있습니다.

      PROJECT=project-id
      

      BUCKET_NAME=bucket-name
      

      CLUSTER=cluster-name
      

      REGION=cluster-region Example: "us-central1"
      

   3. Dataproc 클러스터를 만듭니다. 아래 명령어를 실행하여 지정된 Compute Engine 영역에서 단일 노드 Dataproc 클러스터를 만듭니다.

      gcloud dataproc clusters create ${CLUSTER} \
          --project=${PROJECT} \
          --region=${REGION} \
          --single-node
      

   4. Cloud Storage 버킷에 공개 데이터를 복사합니다. 공개 데이터 Shakespeare 텍스트 스니펫을 Cloud Storage 버킷의 input 폴더에 복사합니다.

      gcloud storage cp gs://pub/shakespeare/rose.txt \
          gs://${BUCKET_NAME}/input/rose.txt
      

   5. 자바(Apache Maven), Scala(SBT), Python 개발 환경을 설정합니다.

   Spark WordCount 작업 준비

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

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

   자바

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

      <?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>

   2. 아래 나열된 WordCount.java 코드를 로컬 머신에 복사합니다.
      1. src/main/java/dataproc/codelab 경로를 사용하여 디렉터리 집합을 만듭니다.

         mkdir -p src/main/java/dataproc/codelab
         

      2. WordCount.java를 로컬 머신의 src/main/java/dataproc/codelab에 복사합니다.

         cp WordCount.java src/main/java/dataproc/codelab
         

      WordCount.java는 Cloud Storage에서 텍스트 파일을 읽고 단어 수를 계산한 후 텍스트 파일 결과를 Cloud Storage에 쓰는 Java의 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);
        }
      }

   3. 패키지를 빌드합니다.

      mvn clean package
      

      성공적으로 빌드되면 target/word-count-1.0.jar이 생성됩니다.
   4. 패키지를 Cloud Storage로 스테이징합니다.

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

   Scala

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

      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"
      )

   2. 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)
        }
      }

   3. 패키지를 빌드합니다.

      sbt clean package
      

      성공적으로 빌드되면 target/scala-2.11/word-count_2.11-1.0.jar이 생성됩니다.
   4. 패키지를 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

   1. 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)
   

   삭제

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

   프로젝트 삭제

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

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

   1. In the Google Cloud console, go to the Manage resources page.

      Go to Manage resources

   2. In the project list, select the project that you want to delete, and then click Delete.
   3. In the dialog, type the project ID, and then click Shut down to delete the project.

   Dataproc 클러스터 삭제

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

   Cloud Storage 버킷 삭제

   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 Delete, and then follow the instructions.

   다음 단계

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