Apache Spark로 Cloud Storage 커넥터 사용

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

Lightning Engine은 Cloud Storage와의 연결을 강화하여 네이티브 엔진의 성능을 최적화합니다. 개선된 Cloud Storage 커넥터는 메타데이터 작업을 최소화하여 비용을 절감하고, 최적화된 파일 출력 커미터는 Spark 워크로드의 성능과 안정성을 높입니다. 사전 체험판 양식을 작성하여 이 비공개 미리보기 기능의 사전 체험판을 요청하세요.

목표

Java, Scala 또는 Python에서 간단한 WordCount Spark 작업을 작성한 다음 Managed Service for Apache Spark 클러스터에서 작업을 실행합니다.

비용

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

  • Compute Engine
  • Managed Service for Apache Spark
  • Cloud Storage

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

Google Cloud 신규 사용자는 무료 체험판을 사용할 수 있습니다.

시작하기 전에

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

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

    1. Google Cloud 계정에 로그인합니다. Google Cloud를 처음 사용하는 경우 계정을 만들고 Google 제품의 실제 성능을 평가해 보세요. 신규 고객에게는 워크로드를 실행, 테스트, 배포하는 데 사용할 수 있는 $300의 무료 크레딧이 제공됩니다.
    2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

      Roles required to select or create a project

      • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
      • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

      Go to project selector

    3. Verify that billing is enabled for your Google Cloud project.

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

      Roles required to enable APIs

      To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

      Enable the APIs

    5. Create a service account:

      1. Ensure that you have the Create Service Accounts IAM role (roles/iam.serviceAccountCreator) and the Project IAM Admin role (roles/resourcemanager.projectIamAdmin). Learn how to grant roles.
      2. In the Google Cloud console, go to the Create service account page.

        Go to Create service account
      3. Select your project.
      4. 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.

      5. Click Create and continue.
      6. Grant the Project > Owner role to the service account.

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

      7. Click Continue.
      8. Click Done to finish creating the service account.

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

    6. 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.
    7. 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.

    8. Google Cloud CLI를 설치합니다.

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

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

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

      Roles required to select or create a project

      • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
      • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

      Go to project selector

    12. Verify that billing is enabled for your Google Cloud project.

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

      Roles required to enable APIs

      To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

      Enable the APIs

    14. Create a service account:

      1. Ensure that you have the Create Service Accounts IAM role (roles/iam.serviceAccountCreator) and the Project IAM Admin role (roles/resourcemanager.projectIamAdmin). Learn how to grant roles.
      2. In the Google Cloud console, go to the Create service account page.

        Go to Create service account
      3. Select your project.
      4. 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.

      5. Click Create and continue.
      6. Grant the Project > Owner role to the service account.

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

      7. Click Continue.
      8. Click Done to finish creating the service account.

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

    15. 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.
    16. 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.

    17. Google Cloud CLI를 설치합니다.

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

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

      gcloud init

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

    1. Google Cloud 콘솔에서 Cloud Storage 버킷 페이지로 이동합니다.

      버킷으로 이동

    2. 만들기를 클릭합니다.
    3. 버킷 만들기 페이지에서 버킷 정보를 입력합니다. 다음 단계로 이동하려면 계속을 클릭합니다.
      1. 시작하기 섹션에서 다음을 수행합니다.
        • 버킷 이름 요구사항을 충족하는 전역적으로 고유한 이름을 입력합니다.
        • 버킷 라벨을 추가하려면 라벨 섹션 ()을 펼치고 라벨 추가를 클릭한 다음 라벨에 keyvalue를 지정합니다.
      2. 데이터 저장 위치 선택 섹션에서 다음을 수행합니다.
        1. 위치 유형을 선택합니다.
        2. 위치 유형 드롭다운 메뉴에서 버킷의 데이터가 영구적으로 저장될 위치를 선택합니다.
          • 이중 리전 위치 유형을 선택하는 경우 관련 체크박스를 사용하여 터보 복제를 사용 설정할 수도 있습니다.
        3. 버킷 간 복제를 설정하려면 Storage Transfer Service를 통해 버킷 간 복제 추가를 선택하고 다음 단계를 따르세요.

          버킷 간 복제 설정

          1. 버킷 메뉴에서 버킷을 선택합니다.
          2. 복제 설정 섹션에서 구성을 클릭하여 복제 작업의 설정을 구성합니다.

            버킷 간 복제 구성 창이 표시됩니다.

            • 객체 이름 프리픽스를 기준으로 복제할 객체를 필터링하려면 객체를 포함하거나 제외할 프리픽스를 입력한 다음 프리픽스 추가를 클릭합니다.
            • 복제된 객체의 스토리지 클래스를 설정하려면 스토리지 클래스 메뉴에서 스토리지 클래스를 선택합니다. 이 단계를 건너뛰면 복제된 객체가 기본적으로 대상 버킷의 스토리지 클래스를 사용합니다.
            • 완료를 클릭합니다.
      3. 데이터 저장 방법 선택 섹션에서 다음을 수행합니다.
        1. 버킷의 기본 스토리지 클래스를 선택하거나 버킷 데이터의 자동 스토리지 클래스 관리에서 자동 클래스를 선택합니다.
        2. 계층적 네임스페이스를 사용 설정하려면 데이터 집약적인 워크로드에 맞게 스토리지 최적화 섹션에서 이 버킷에서 계층적 네임스페이스 사용 설정을 선택합니다.
      4. 객체 액세스를 제어하는 방식 선택 섹션에서 버킷이 공개 액세스 방지를 적용할지 여부를 선택하고 버킷의 객체에 대한 액세스 제어 방법을 선택합니다.
      5. 객체 데이터 보호 방법 선택 섹션에서 다음을 수행합니다.
        • 버킷에 설정할 데이터 보호 아래의 옵션을 선택합니다.
          • 소프트 삭제를 사용 설정하려면 소프트 삭제 정책 (데이터 복구용) 체크박스를 클릭하고 삭제 후 객체를 보관하려는 일수를 지정합니다.
          • 객체 버전 관리를 설정하려면 객체 버전 관리 (버전 제어용) 체크박스를 클릭하고 객체당 최대 버전 수와 이전 버전이 만료되는 일수를 지정합니다.
          • 객체 및 버킷에 보관 정책을 사용 설정하려면 보관 (규정 준수용) 체크박스를 클릭한 후 다음 단계를 따르세요.
            • 객체 보관 잠금을 사용 설정하려면 객체 보관 사용 설정 체크박스를 클릭합니다.
            • 버킷 잠금을 사용 설정하려면 버킷 보관 정책 설정 체크박스를 클릭하고 보관 기간의 단위와 기간을 선택합니다.
        • 객체 데이터가 암호화되는 방식을 선택하려면 데이터 암호화 섹션 ()을 펼치고 데이터 암호화 방법을 선택합니다.
    4. 만들기를 클릭합니다.

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

    PROJECT=project-id
    
    BUCKET_NAME=bucket-name
    
    CLUSTER=cluster-name
    
    REGION=cluster-region Example: "us-central1"
    

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

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

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

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

  6. Java(Apache Maven), Scala(SBT), Python 개발 환경을 설정합니다.

Spark WordCount 작업 준비

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

자바

  1. pom.xml 파일을 로컬 머신에 복사합니다. 다음 pom.xml 파일은 Scala 및 Spark 라이브러리 종속 항목을 지정합니다. 이는 Managed Service for Apache Spark 클러스터가 런타임 시 이러한 라이브러리를 제공함을 나타내기 위해 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/managed-spark/codelab 경로를 사용하여 디렉터리 집합을 만듭니다.
      mkdir -p src/main/java/managed-spark/codelab
      
    2. WordCount.java를 로컬 머신의 src/main/java/managed-spark/codelab에 복사합니다.
      cp WordCount.java src/main/java/managed-spark/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 라이브러리 종속 항목을 지정합니다. 이는 Managed Service for Apache Spark 클러스터가 런타임 시 이러한 라이브러리를 제공함을 나타내기 위해 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 작업을 Managed Service for Apache Spark 클러스터에 제출합니다.

자바

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. Google Cloud 콘솔에서 리소스 관리 페이지로 이동합니다.

    리소스 관리로 이동

  2. 프로젝트 목록에서 삭제할 프로젝트를 선택하고 삭제를 클릭합니다.
  3. 대화상자에서 프로젝트 ID를 입력한 후 종료를 클릭하여 프로젝트를 삭제합니다.

Managed Service for Apache Spark 클러스터 삭제

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

Cloud Storage 버킷 삭제

Google Cloud 콘솔

  1. Google Cloud 콘솔에서 Cloud Storage 버킷 페이지로 이동합니다.

    버킷으로 이동

  2. 삭제할 버킷의 체크박스를 클릭합니다.
  3. 버킷을 삭제하려면 삭제를 클릭한 후 안내를 따르세요.

명령줄

    버킷을 삭제합니다.
    gcloud storage buckets delete BUCKET_NAME

다음 단계