Utilizzo del connettore Cloud Storage con Apache Spark


Questo tutorial mostra come eseguire codice di esempio che utilizza il connettore Cloud Storage con Apache Spark.

Obiettivi

Scrivi un semplice job Spark per il conteggio di parole in Java, Scala o Python, quindi esegui il job su un cluster Dataproc.

Costi

In questo documento, utilizzi i seguenti componenti fatturabili di Google Cloud:

  • Compute Engine
  • Dataproc
  • Cloud Storage

Per generare una stima dei costi in base all'utilizzo previsto, utilizza il Calcolatore prezzi. I nuovi utenti di Google Cloud potrebbero essere idonei per una prova gratuita.

Prima di iniziare

Esegui i passaggi riportati di seguito per prepararti a eseguire il codice in questo tutorial.

  1. Configura il tuo progetto. Se necessario, configura un progetto con le API Dataproc, Compute Engine e Cloud Storage abilitate e Google Cloud CLI installato sulla tua macchina locale.

    1. Accedi al tuo account Google Cloud. Se non conosci Google Cloud, crea un account per valutare le prestazioni dei nostri prodotti in scenari reali. I nuovi clienti ricevono anche 300 $di crediti gratuiti per l'esecuzione, il test e il deployment dei carichi di lavoro.
    2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

      Go to project selector

    3. Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.

    4. Abilita le API Dataproc, Compute Engine, and Cloud Storage.

      Abilita le API

    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. Imposta la variabile di ambiente GOOGLE_APPLICATION_CREDENTIALS sul percorso del file JSON che contiene le tue credenziali. Questa variabile si applica solo alla sessione di shell attuale. Pertanto, se apri una nuova sessione, imposta di nuovo la variabile.

    8. Installa Google Cloud CLI.
    9. Per initialize gcloud CLI, esegui questo comando:

      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. Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.

    12. Abilita le API Dataproc, Compute Engine, and Cloud Storage.

      Abilita le API

    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. Imposta la variabile di ambiente GOOGLE_APPLICATION_CREDENTIALS sul percorso del file JSON che contiene le tue credenziali. Questa variabile si applica solo alla sessione di shell attuale. Pertanto, se apri una nuova sessione, imposta di nuovo la variabile.

    16. Installa Google Cloud CLI.
    17. Per initialize gcloud CLI, esegui questo comando:

      gcloud init

  2. Crea un bucket Cloud Storage. È necessario un servizio Cloud Storage per archiviare i dati del tutorial. Se non ne hai uno pronto per l'uso, crea un nuovo bucket nel progetto.

    1. Nella console Google Cloud, vai alla pagina Bucket di Cloud Storage.

      Vai alla pagina Bucket

    2. Fai clic su Crea bucket.
    3. Nella pagina Crea un bucket, inserisci le informazioni del bucket. Per andare al passaggio successivo, fai clic su Continua.
    4. Fai clic su Crea.

  3. Imposta le variabili di ambiente locali. Imposta le variabili di ambiente sulla tua macchina locale. Imposta l'ID progetto Google Cloud e il nome del bucket Cloud Storage che utilizzerai per questo tutorial. Fornisci anche il nome e la regione di un cluster Dataproc nuovo o esistente. Puoi creare un cluster da utilizzare in questo tutorial nel passaggio successivo.

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

  4. Crea un cluster Dataproc. Esegui il comando riportato di seguito per creare un cluster Dataproc a nodo singolo nella zona di Compute Engine specificata.

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

  5. Copia i dati pubblici nel bucket Cloud Storage. Copia uno snippet di testo di dati pubblici nella cartella input del bucket Cloud Storage:

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

  6. Configura un ambiente di sviluppo Java (Apache Maven), Scala (SBT) o Python.

Prepara il job di conteggio parole Spark

Seleziona una scheda di seguito per seguire i passaggi per preparare un pacchetto o un file job da inviare al cluster. Puoi preparare uno dei seguenti tipi di prestazioni:

Java

  1. Copia pom.xml file sulla tua macchina locale. Il seguente file pom.xml specifica le dipendenze delle librerie Scala e Spark a cui è assegnato un ambito provided per indicare che il cluster Dataproc fornirà queste librerie in fase di runtime. Il file pom.xml non specifica una dipendenza Cloud Storage perché il connettore implementa l'interfaccia HDFS standard. Quando un job Spark accede ai file del cluster Cloud Storage (file con URI che iniziano con gs://), il sistema utilizza automaticamente il connettore Cloud Storage per accedere ai file in 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. Copia il codice WordCount.java elencato di seguito sulla tua macchina locale.
    1. Crea un insieme di directory con il percorso src/main/java/dataproc/codelab:
      mkdir -p src/main/java/dataproc/codelab
      
    2. Copia WordCount.java sulla tua macchina locale in src/main/java/dataproc/codelab:
      cp WordCount.java src/main/java/dataproc/codelab
      

    WordCount.java è un semplice job Spark in Java che legge i file di testo da Cloud Storage, esegue il conteggio delle parole e scrive i risultati dei file di testo in 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. Crea il pacchetto.
    mvn clean package
    
    Se la build ha esito positivo, viene creato un target/word-count-1.0.jar.
  4. Esegui l'archiviazione temporanea del pacchetto in Cloud Storage.
    gsutil cp target/word-count-1.0.jar \
        gs://${BUCKET_NAME}/java/word-count-1.0.jar
    

Scala

  1. Copia build.sbt file sulla tua macchina locale. Il seguente file build.sbt specifica le dipendenze delle librerie Scala e Spark a cui è assegnato un ambito provided per indicare che il cluster Dataproc fornirà queste librerie in fase di runtime. Il file build.sbt non specifica una dipendenza Cloud Storage perché il connettore implementa l'interfaccia HDFS standard. Quando un job Spark accede ai file del cluster Cloud Storage (file con URI che iniziano con gs://), il sistema utilizza automaticamente il connettore Cloud Storage per accedere ai file in 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. Copia word-count.scala sul tuo computer locale. Si tratta di un semplice job Spark in Java che legge i file di testo da Cloud Storage, esegue il conteggio delle parole e scrive i risultati del file di testo in 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. Crea il pacchetto.
    sbt clean package
    
    Se la build ha esito positivo, viene creato un target/scala-2.11/word-count_2.11-1.0.jar.
  4. Esegui l'archiviazione temporanea del pacchetto in Cloud Storage.
    gsutil 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. Copia word-count.py sul tuo computer locale. Si tratta di un semplice job Spark in Python che utilizza PySpark che legge i file di testo da Cloud Storage, esegue un conteggio delle parole e scrive i risultati dei file di testo in 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])
    

Invia il job

Esegui questo comando gcloud per inviare il job di conteggio parole al cluster 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/

Visualizza l'output

Al termine del job, esegui il seguente comando gsutil dell'interfaccia a riga di comando gcloud per visualizzare l'output del conteggio parole.

gsutil cat gs://${BUCKET_NAME}/output/*

L'output del conteggio parole dovrebbe essere simile al seguente:

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

Esegui la pulizia

Al termine del tutorial, puoi eseguire la pulizia delle risorse che hai creato in modo che smettano di utilizzare la quota e non incorrano in addebiti. Le sezioni seguenti descrivono come eliminare o disattivare queste risorse.

Elimina il progetto

Il modo più semplice per eliminare la fatturazione è eliminare il progetto che hai creato per il tutorial.

Per eliminare il progetto:

  1. Nella console Google Cloud, vai alla pagina Gestisci risorse.

    Vai a Gestisci risorse

  2. Nell'elenco dei progetti, seleziona il progetto che vuoi eliminare, quindi fai clic su Elimina.
  3. Nella finestra di dialogo, digita l'ID del progetto e fai clic su Chiudi per eliminare il progetto.

Eliminazione del cluster Dataproc

Invece di eliminare il progetto, potresti voler eliminare il cluster solo all'interno del progetto.

Eliminazione del bucket Cloud Storage

Console Google Cloud

  1. Nella console Google Cloud, vai alla pagina Bucket di Cloud Storage.

    Vai a Bucket

  2. Fai clic sulla casella di controllo relativa al bucket da eliminare.
  3. Per eliminare il bucket, fai clic su Elimina e segui le istruzioni.

Riga di comando

    Elimina il bucket:
    gcloud storage buckets delete BUCKET_NAME

Passaggi successivi