Cloud Storage-Connector mit Apache Spark verwenden


In dieser Anleitung erfahren Sie, wie Sie Beispielcode ausführen, der den Cloud Storage-Connector mit Apache Spark verwendet.

Lernziele

Schreiben Sie einen einfachen Wordcount-Spark-Job in Java, Scala oder Python und führen Sie den Job dann in einem Dataproc-Cluster aus.

Kosten

In diesem Dokument verwenden Sie die folgenden kostenpflichtigen Komponenten von Google Cloud:

  • Compute Engine
  • Dataproc
  • Cloud Storage

Mit dem Preisrechner können Sie eine Kostenschätzung für Ihre voraussichtliche Nutzung vornehmen. Neuen Google Cloud-Nutzern steht möglicherweise eine kostenlose Testversion zur Verfügung.

Hinweise

Führen Sie die Schritte unten aus, um den Code für diese Anleitung vorzubereiten.

  1. Richten Sie ein Projekt ein. Richten Sie bei Bedarf ein Projekt mit Dataproc, Compute Engine aktiviert sind und die Google Cloud CLI auf Ihrem lokalen Computer installiert ist.

    1. 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.
    2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

      Go to project selector

    3. Make sure that billing is enabled for your Google Cloud project.

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

      Enable the APIs

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

    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. Install the Google Cloud CLI.
    9. To initialize the gcloud CLI, run the following command:

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

      Go to project selector

    11. Make sure that billing is enabled for your Google Cloud project.

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

      Enable the APIs

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

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

    16. Install the Google Cloud CLI.
    17. To initialize the gcloud CLI, run the following command:

      gcloud init

  2. Erstellen Sie einen Cloud Storage-Bucket. Sie benötigen Cloud Storage, um Anleitungsdaten zu speichern. Wenn Sie den Dienst nicht haben, können Sie einen neuen Bucket in Ihrem Projekt erstellen.

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

      Go to Buckets page

    2. Click Create bucket.
    3. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
      • For Name your bucket, enter a name that meets the bucket naming requirements.
      • For Choose where to store your data, do the following:
        • Select a Location type option.
        • Select a Location option.
      • For Choose a default storage class for your data, select a storage class.
      • For Choose how to control access to objects, select an Access control option.
      • For Advanced settings (optional), specify an encryption method, a retention policy, or bucket labels.
    4. Click Create.

  3. Legen Sie lokale Umgebungsvariablen fest. Legen Sie Umgebungsvariablen auf Ihrem lokalen Computer fest. Legen Sie Ihre Google Cloud-Projekt-ID und den Namen des Cloud Storage-Buckets fest, den Sie für diese Anleitung verwenden werden. Geben Sie außerdem den Namen und die Region eines vorhandenen oder neuen Dataproc-Clusters an. Sie können im nächsten Schritt einen Cluster erstellen, der in dieser Anleitung verwendet werden soll.

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

  4. Dataproc-Cluster erstellen. Führen Sie den folgenden Befehl aus, um einen Dataproc-Cluster mit einem einzelnen Knoten in der angegebenen Compute Engine-Zone zu erstellen.

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

  5. Kopieren Sie öffentliche Daten in Ihren Cloud Storage-Bucket. Kopieren Sie ein öffentliches Data-Shakespeare-Text-Snippet in den Ordner input Ihres Cloud Storage-Buckets:

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

  6. Java- (Apache Maven), Scala- (SBT) oder Python- Entwicklungsumgebung einrichten.

Spark-WordCount-Job vorbereiten

Wählen Sie unten einen Tab aus, um die Schritte zum Vorbereiten eines Jobpakets oder einer Datei zum Senden an den Cluster auszuführen. Sie können einen der folgenden Jobtypen vorbereiten:

Java

  1. pom.xml-Datei auf Ihren lokalen Computer kopieren. Die folgende pom.xml-Datei gibt Scala- und Spark-Bibliotheksabhängigkeiten an, denen ein provided-Bereich zugewiesen wird, um anzugeben, dass der Dataproc-Cluster diese Bibliotheken zur Laufzeit bereitstellt. Die pom.xml-Datei gibt keine Cloud Storage-Abhängigkeit an, da der Connector die standardmäßige HDFS-Schnittstelle implementiert. Wenn ein Spark-Job auf Cloud Storage-Clusterdateien zugreift (Dateien mit URIs, die mit gs:// beginnen), verwendet das System automatisch den Cloud Storage-Connector, um auf die Dateien in Cloud Storage zuzugreifen
    <?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. Den unten aufgeführten WordCount.java-Code auf Ihren lokalen Computer kopieren.
    1. Erstellen Sie einen Satz von Verzeichnissen mit dem Pfad src/main/java/dataproc/codelab:
      mkdir -p src/main/java/dataproc/codelab
      
    2. Kopieren Sie WordCount.java auf Ihren lokalen Computer in src/main/java/dataproc/codelab:
      cp WordCount.java src/main/java/dataproc/codelab
      

    WordCount.java ist ein einfacher Spark-Job in Java, der Textdateien aus Cloud Storage liest, eine Wortzählung durchführt und dann die Ergebnisse der Textdatei in Cloud Storage schreibt.

    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. Erstellen Sie das Paket.
    mvn clean package
    
    Ist dies erfolgreich, wird eine target/word-count-1.0.jar-Datei erstellt.
  4. Das Paket in Cloud Storage bereitstellen.
    gcloud storage cp target/word-count-1.0.jar \
        gs://${BUCKET_NAME}/java/word-count-1.0.jar
    

Scala

  1. build.sbt-Datei auf Ihren lokalen Computer kopieren. Die folgende build.sbt-Datei gibt Scala- und Spark-Bibliotheksabhängigkeiten an, denen ein provided-Bereich zugewiesen wird, um anzugeben, dass der Dataproc-Cluster diese Bibliotheken zur Laufzeit bereitstellt. Die build.sbt-Datei gibt keine Cloud Storage-Abhängigkeit an, da der Connector die standardmäßige HDFS-Schnittstelle implementiert. Wenn ein Spark-Job auf Cloud Storage-Clusterdateien zugreift (Dateien mit URIs, die mit gs:// beginnen), verwendet das System automatisch den Cloud Storage-Connector, um auf die Dateien in Cloud Storage zuzugreifen
    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 auf Ihren lokalen Computer kopieren. Dies ist ein einfacher Spark-Job in Java, der Textdateien aus Cloud Storage liest, eine Wortzählung ausführt und die Ergebnisse der Textdatei anschließend in Cloud Storage schreibt.
    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. Erstellen Sie das Paket.
    sbt clean package
    
    Ist dies erfolgreich, wird eine target/scala-2.11/word-count_2.11-1.0.jar-Datei erstellt.
  4. Das Paket in Cloud Storage bereitstellen.
    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 auf Ihren lokalen Computer kopieren. Dies ist ein einfacher Spark-Job in Python mit PySpark, der Textdateien aus Cloud Storage liest, eine Wortzählung ausführt und die Ergebnisse der Textdatei anschließend in Cloud Storage schreibt.
    #!/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])

Job senden

Führen Sie den folgenden gcloud-Befehl aus, um den Wordcount-Job an Ihren Dataproc-Cluster zu senden.

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/

Ausgabe ansehen

Führen Sie nach Abschluss des Jobs die folgende gcloud CLI aus um die Wordcount-Ausgabe anzuzeigen.

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

Es sollte etwa diese Ausgabe angezeigt werden:

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

Bereinigen

Nachdem Sie die Anleitung abgeschlossen haben, können Sie die erstellten Ressourcen bereinigen, damit sie keine Kontingente mehr nutzen und keine Gebühren mehr anfallen. In den folgenden Abschnitten erfahren Sie, wie Sie diese Ressourcen löschen oder deaktivieren.

Projekt löschen

Am einfachsten vermeiden Sie weitere Kosten, wenn Sie das zum Ausführen der Anleitung erstellte Projekt löschen.

So löschen Sie das Projekt:

  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-Cluster löschen

Wenn Sie Ihr Projekt löschen möchten, sollten Sie nur Ihren Cluster innerhalb des Projekts löschen.

Cloud Storage-Bucket löschen

Google Cloud Console

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

    Go to Buckets

  2. Click the checkbox for the bucket that you want to delete.
  3. To delete the bucket, click Delete, and then follow the instructions.

Befehlszeile

    Löschen Sie den Bucket:
    gcloud storage buckets delete BUCKET_NAME

Nächste Schritte