搭配 Apache Spark 使用 Cloud Storage 連接器

---

本教學課程說明如何執行程式碼範例，搭配使用 Cloud Storage 連接器和 Apache Spark。

目標

以 Java、Scala 或 Python 編寫簡單的字數計算 Spark 工作，然後在 Dataproc 叢集上執行該工作。

所有 Dataproc 叢集映像檔版本都已安裝本教學課程所需的 Spark 元件。

費用

在本文件中，您會使用 Google Cloud的下列計費元件：

• Compute Engine
• Dataproc
• Cloud Storage

如要根據預測用量估算費用，請使用 Pricing Calculator。

初次使用 Google Cloud 的使用者可能符合免費試用資格。

事前準備

請按照下列步驟操作，準備在本教學課程中執行程式碼。

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.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the 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.

      Note: The Role field affects which resources the service account can access in your project. You can revoke these roles or grant additional roles later. In production environments, do not grant the Owner, Editor, or Viewer roles. Instead, grant a predefined role or custom role that meets your needs.
   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.

   如果您使用外部識別資訊提供者 (IdP)，請先 使用聯合身分登入 gcloud CLI。

   如要初始化 gcloud CLI，請執行下列指令：

   gcloud init

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

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the 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.

      Note: The Role field affects which resources the service account can access in your project. You can revoke these roles or grant additional roles later. In production environments, do not grant the Owner, Editor, or Viewer roles. Instead, grant a predefined role or custom role that meets your needs.
   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.

   如果您使用外部識別資訊提供者 (IdP)，請先 使用聯合身分登入 gcloud CLI。

   如要初始化 gcloud CLI，請執行下列指令：

   gcloud init

   1. 建立 Cloud Storage bucket。您需要 Cloud Storage 來保存教學課程資料。如果沒有可用的值區，請在專案中建立新值區。

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

         Go to Buckets

      2. Click 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 (), click 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 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.
               Note: You cannot enable hierarchical namespace in existing buckets.
         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.
            Note: You cannot change the Prevent public access setting if this setting is enforced at an organization policy.
         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 (), and select a Data encryption method.
      4. Click Create.

   2. 設定本機環境變數。在本機上設定環境變數。設定 Google Cloud project-id 和您在本教學課程中使用的 Cloud Storage 值區名稱。此外，請提供現有或新的 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
      

      上述指令會安裝預設的叢集映像檔版本。 您可以使用 --image-version 旗標，為叢集選取映像檔版本。每個映像檔版本都會安裝特定版本的 Spark 和 Scala 程式庫元件。如果您準備以 Java 或 Scala 執行 Spark 字數統計工作，準備工作套件時，請參照叢集上安裝的 Spark 和 Scala 版本。

   4. 將公開資料複製到 Cloud Storage 值區。將莎士比亞文字片段的公開資料複製到 Cloud Storage bucket 的 input 資料夾：

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

   5. 設定 Java (Apache Maven)、Scala (SBT) 或 Python 開發環境。

      使用 Cloud Shell。 Cloud Shell 內含本教學課程使用的工具，包括 Apache Maven、Python 和 Google Cloud CLI。

   準備 Spark 字數統計工作

   選取下方分頁標籤，按照步驟準備要提交至叢集的工作套件或檔案。你可以準備下列其中一種工作類型：

   • 使用 Java 執行 Spark 工作： 使用 Apache Maven 建構 JAR 封裝
   • 使用 SBT 建構 JAR 套件的 Scala 中的 Spark 工作
   • Python 中的 Spark 工作 (PySpark)

   Java

   1. 將 pom.xml 檔案複製到本機電腦。 下列 pom.xml 檔案會指定 Scala 和 Spark 程式庫依附元件，這些依附元件會獲得 provided 範圍，表示 Dataproc 叢集會在執行階段提供這些程式庫。pom.xml 檔案不會指定 Cloud Storage 依附元件，因為連接器會實作標準 HDFS 介面。當 Spark 工作存取 Cloud Storage 叢集檔案 (URI 開頭為 gs:// 的檔案) 時，系統會自動使用 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>

   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 是以 Java 執行的 Spark 工作，可從 Cloud Storage 讀取文字檔案、執行字數統計，然後將文字檔案結果寫入 Cloud Storage。

      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 程式庫依附元件，這些依附元件會獲得 provided 範圍，表示 Dataproc 叢集會在執行階段提供這些程式庫。build.sbt 檔案不會指定 Cloud Storage 依附元件，因為連接器會實作標準 HDFS 介面。當 Spark 工作存取 Cloud Storage 叢集檔案 (URI 開頭為 gs:// 的檔案) 時，系統會自動使用 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"
      )

   2. 將 word-count.scala 複製到本機電腦。 這是以 Java 執行的 Spark 工作，可從 Cloud Storage 讀取文字檔案、執行字數統計，然後將文字檔案結果寫入 Cloud Storage。

      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 的 Python Spark 工作，可從 Cloud Storage 讀取文字檔案、執行字數統計，然後將文字檔案結果寫入 Cloud Storage。

      #!/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 叢集。

   Java

   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 指令，查看字數統計輸出內容。

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

   字數統計輸出內容應如下所示：

   (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。如要保留使用該專案 ID 的網址 (例如 appspot.com 網址)，請刪除在該專案中選取的資源，而不是刪除整個專案。

   如果打算探索多種架構、教學課程或快速入門導覽課程，重複使用專案有助於避免超出專案配額限制。

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

   後續步驟

   • 請參閱 Spark 工作調整提示