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.

Ziele

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 Nutzern von Google Cloud steht möglicherweise eine kostenlose Testversion zur Verfügung.

Hinweis

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

Richten Sie ein Projekt ein. Richten Sie bei Bedarf ein Projekt mit aktivierten Dataproc-, Compute Engine- und Cloud Storage-APIs ein und installieren Sie die Google Cloud CLI auf Ihrem lokalen Computer.
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. Wechseln Sie in der Google Cloud Console unter „Cloud Storage“ zur Seite Buckets.
  Buckets aufrufen
2. Klicken Sie auf Erstellen.
3. Geben Sie auf der Seite Bucket erstellen die Bucket-Informationen ein. Klicken Sie auf Weiter, um mit dem nächsten Schritt fortzufahren.
  1. Führen Sie im Abschnitt Einstieg die folgenden Schritte aus:
    - Geben Sie einen global eindeutigen Namen ein, der den Anforderungen für Bucket-Namen entspricht.
    - So fügen Sie ein Bucket-Label hinzu: Maximieren Sie den Abschnitt Labels (), klicken Sie auf Label hinzufügen und geben Sie key und value für Ihr Label an.
  2. Gehen Sie im Bereich Speicherort für Daten auswählen so vor:
    1. Standorttyp auswählen.
    2. Wählen Sie im Drop-down-Menü Standorttyp einen Standort aus, an dem die Daten Ihres Buckets dauerhaft gespeichert werden.
      - Wenn Sie den Standorttyp Dual-Region auswählen, können Sie auch die Turboreplikation aktivieren, indem Sie das entsprechende Kästchen anklicken.
    3. Wenn Sie die Bucket-übergreifende Replikation einrichten möchten, wählen Sie Bucket-übergreifende Replikation über Storage Transfer Service hinzufügen aus und führen Sie die folgenden Schritte aus:
      Bucket-übergreifende Replikation einrichten
      
      Wählen Sie im Menü Bucket einen Bucket aus.
      
      Klicken Sie im Bereich Replikationseinstellungen auf Konfigurieren, um die Einstellungen für den Replikationsjob zu konfigurieren.
      
      Der Bereich Bucket-übergreifende Replikation konfigurieren wird angezeigt.
      
      Wenn Sie die zu replizierenden Objekte nach dem Objektnamenspräfix filtern möchten, geben Sie ein Präfix ein, mit dem Sie Objekte ein- oder ausschließen möchten, und klicken Sie dann auf Präfix hinzufügen.
      
      Wenn Sie eine Speicherklasse für die replizierten Objekte festlegen möchten, wählen Sie im Menü Speicherklasse eine Speicherklasse aus. Wenn Sie diesen Schritt überspringen, wird für replizierte Objekte standardmäßig die Speicherklasse des Ziel-Buckets verwendet.
      
      Klicken Sie auf Fertig.
  3. Gehen Sie im Bereich Speicherort für Daten auswählen so vor:
    1. Wählen Sie eine Standardspeicherklasse für den Bucket oder Autoclass für die automatische Speicherklassenverwaltung der Daten Ihres Buckets aus.
    2. Wenn Sie den hierarchischen Namespace aktivieren möchten, wählen Sie im Bereich Speicher für datenintensive Arbeitslasten optimieren die Option Hierarchischen Namespace für diesen Bucket aktivieren aus.
      Hinweis:Sie können den hierarchischen Namespace nicht in vorhandenen Buckets aktivieren.
  4. Wählen Sie im Abschnitt Zugriff auf Objekte steuern aus, ob der Bucket Verhinderung des öffentlichen Zugriffs durchsetzt, und wählen Sie eine Zugriffssteuerungsmethode für die Objekte Ihres Buckets aus.
    Hinweis:Sie können die Einstellung Öffentlichen Zugriff verhindern nicht ändern, wenn sie durch eine Organisationsrichtlinie erzwungen wird.
  5. Führen Sie im Bereich Auswählen, wie Objektdaten geschützt werden die folgenden Schritte aus:
    - Wählen Sie unter Datenschutz die gewünschten Optionen für Ihren Bucket aus.
      - Wenn Sie Vorläufiges Löschen aktivieren möchten, klicken Sie das Kästchen Richtlinie für vorläufiges Löschen (zur Datenwiederherstellung) an und geben Sie die Anzahl der Tage an, die Objekte nach dem Löschen beibehalten werden sollen.
      - Wenn Sie die Objektversionsverwaltung festlegen möchten, klicken Sie das Kästchen Objektversionsverwaltung (zur Datenwiederherstellung) an und geben Sie die maximale Anzahl von Versionen pro Objekt und die Anzahl der Tage an, nach denen die nicht aktuellen Versionen ablaufen.
      - Klicken Sie das Kästchen Aufbewahrung (für Compliance) an, um die Aufbewahrungsrichtlinie für Objekte und Buckets zu aktivieren, und gehen Sie dann so vor:
        
        Klicken Sie auf das Kästchen Objektaufbewahrung aktivieren, um die Objektaufbewahrungssperre zu aktivieren.
        
        Wenn Sie Bucket Lock aktivieren möchten, klicken Sie das Kästchen Bucket-Aufbewahrungsrichtlinie festlegen an und wählen Sie eine Zeiteinheit und eine Zeitdauer für die Aufbewahrungsdauer aus.
    - Um auszuwählen, wie Ihre Objektdaten verschlüsselt werden, maximieren Sie den Bereich Datenverschlüsselung () und wählen Sie eine Methode für die Datenverschlüsselung aus.
4. Klicken Sie auf Erstellen.
Legen Sie lokale Umgebungsvariablen fest. Legen Sie Umgebungsvariablen auf Ihrem lokalen Computer fest. Legen Sie Ihre Google Cloud project-id und den Namen des Cloud Storage-Bucket 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"
```
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
```
Mit dem obigen Befehl wird die standardmäßige Cluster-Image-Version installiert. Mit dem Flag --image-version können Sie eine Image-Version für Ihren Cluster auswählen. Jede Image-Version installiert bestimmte Versionen von Spark- und Scala-Bibliothekskomponenten. Wenn Sie den Spark-Wordcount-Job in Java oder Scala vorbereiten, verweisen Sie auf die Spark- und Scala-Versionen auf Ihrem Cluster, wenn Sie das Jobpaket vorbereiten.
Kopieren Sie öffentliche Daten in Ihren Cloud Storage-Bucket. Kopieren Sie ein öffentliches Data-Shakepee-Text-Snippet in den Ordner input Ihres Cloud Storage-Buckets:
```
gcloud storage cp gs://pub/shakespeare/rose.txt \
    gs://${BUCKET_NAME}/input/rose.txt
```
Java- (Apache Maven), Scala- (SBT) oder Python- Entwicklungsumgebung einrichten.
Cloud Shell verwenden. Cloud Shell enthält in dieser Anleitung verwendete Tools wie Apache Maven, Python und die Google Cloud CLI.

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:

Spark-Job in Java mit Apache Maven zum Erstellen eines JAR-Pakets
Spark-Job in Scala mit SBT zum Erstellen eines JAR-Pakets
Spark-Job in Python (PySpark)

Java

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

Prüfen Sie die Version des Cluster-Images. Ersetzen Sie die Platzhalter version in der Datei, um die von der Image-Version Ihres Clusters verwendeten Spark- und Scala-Bibliotheksversionen anzuzeigen. Beachten Sie, dass die Artefaktnummer spark-core_ die Scala major.minor-Versionsnummer ist.

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

Den unten aufgeführten WordCount.java-Code auf Ihren lokalen Computer kopieren.

Erstellen Sie einen Satz von Verzeichnissen mit dem Pfad src/main/java/dataproc/codelab:
```
mkdir -p src/main/java/dataproc/codelab
```
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 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);
  }
}

Erstellen Sie das Paket.
```
mvn clean package
```
Ist dies erfolgreich, wird eine target/word-count-1.0.jar-Datei erstellt.

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

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
Prüfen Sie die Version des Cluster-Images. Ersetzen Sie die Platzhalter version in der Datei, um die von der Image-Version Ihres Clusters verwendeten Spark- und Scala-Bibliotheksversionen anzuzeigen.
```
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 auf Ihren lokalen Computer kopieren. Dies ist ein 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)
  }
}

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.

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

word-count.py auf Ihren lokalen Computer kopieren. Dies ist ein 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 den folgenden gcloud CLI-Befehl aus, um die Wordcount-Ausgabe aufzurufen.

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 durch Löschen des für die Anleitung erstellten Projekts.

So löschen Sie das Projekt:

Achtung: Das Löschen von Projekten hat folgende Auswirkungen:

Alle Inhalte des Projekts werden gelöscht. Wenn Sie für die Aufgaben in diesem Dokument ein bereits bestehendes Projekt verwendet haben und dieses löschen, werden auch alle anderen im Rahmen des Projekts erstellten Daten gelöscht.
Benutzerdefinierte Projekt-IDs gehen verloren. Beim Erstellen dieses Projekts haben Sie möglicherweise eine benutzerdefinierte Projekt-ID erstellt, die Sie weiterhin verwenden möchten. Damit die URLs, die die Projekt-ID nutzen, zum Beispiel eine appspot.com-URL, erhalten bleiben, sollten Sie ausgewählte Ressourcen innerhalb des Projekts löschen, anstatt das gesamte Projekt.

Wenn Sie mehrere Architekturen, Anleitungen und Kurzanleitungen durcharbeiten möchten, können Sie die Überschreitung von Projektkontingenten verhindern, indem Sie Projekte wiederverwenden.

Wechseln Sie in der Google Cloud -Console zur Seite Ressourcen verwalten.
Zur Seite „Ressourcen verwalten“
Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie dann auf Löschen.
Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Shut down (Beenden), um das Projekt zu löschen.

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

Wechseln Sie in der Google Cloud Console unter „Cloud Storage“ zur Seite Buckets.
Buckets aufrufen
Klicken Sie auf das Kästchen neben dem Bucket, der gelöscht werden soll.
Klicken Sie zum Löschen des Buckets auf Löschen und folgen Sie der Anleitung.

Befehlszeile

gcloud storage buckets delete BUCKET_NAME

Nächste Schritte

Siehe Tipps zur Feinabstimmung von Spark-Jobs

Cloud Storage-Connector mit Apache Spark verwenden Mit Sammlungen den Überblick behalten Sie können Inhalte basierend auf Ihren Einstellungen speichern und kategorisieren.

Ziele

Kosten

Hinweis

Bucket-übergreifende Replikation einrichten

Spark-WordCount-Job vorbereiten

Java

Scala

Python

Job senden

Java

Scala

Python

Ausgabe ansehen

Bereinigen

Projekt löschen

Dataproc-Cluster löschen

Cloud Storage-Bucket löschen

Google Cloud Console

Befehlszeile

Nächste Schritte

Cloud Storage-Connector mit Apache Spark verwenden