Utilizzo del connettore Cloud Storage con Apache Spark

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

Obiettivi

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

Costi

In questo documento vengono utilizzati 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 avere diritto a una prova senza costi.

Prima di iniziare

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

Configura il progetto. Se necessario, configura un progetto con le API Dataproc, Compute Engine e Cloud Storage abilitate e Google Cloud CLI installata sulla tua macchina locale.
Crea un bucket Cloud Storage. Ti serve un bucket Cloud Storage per archiviare i dati del tutorial. Se non ne hai uno pronto all'uso, crea un nuovo bucket nel tuo progetto.
1. Nella console Google Cloud , vai alla pagina Bucket in Cloud Storage.
  Vai a Bucket
2. Fai clic su Crea.
3. Nella pagina Crea un bucket, inserisci le informazioni del bucket. Per andare al passaggio successivo, fai clic su Continua.
  1. Nella sezione Inizia, segui questi passaggi:
    - Inserisci un nome univoco globale che soddisfi i requisiti per la denominazione dei bucket.
    - Per aggiungere un'etichetta bucket, espandi la sezione Etichette (), fai clic su Aggiungi etichetta e specifica un key e un value per l'etichetta.
  2. Nella sezione Scegli dove archiviare i tuoi dati, segui questi passaggi:
    1. Seleziona un Tipo di località.
    2. Scegli una posizione in cui i dati del bucket vengono archiviati in modo permanente dal menu a discesa Tipo di località.
      - Se selezioni il tipo di località a doppia regione, puoi anche scegliere di attivare la replica turbo utilizzando la casella di controllo pertinente.
    3. Per configurare la replica tra bucket, seleziona Aggiungi una replica tra bucket mediante Storage Transfer Service e segui questi passaggi:
      Configura la replica tra bucket
      
      Nel menu Bucket, seleziona un bucket.
      
      Nella sezione Impostazioni di replica, fai clic su Configura per configurare le impostazioni per il job di replica.
      
      Viene visualizzato il riquadro Configura replica tra bucket.
      
      Per filtrare gli oggetti da replicare in base al prefisso del nome dell'oggetto, inserisci un prefisso da cui includere o escludere gli oggetti, quindi fai clic su Aggiungi un prefisso.
      
      Per impostare una classe di archiviazione per gli oggetti replicati, seleziona una classe di archiviazione dal menu Classe di archiviazione. Se salti questo passaggio, gli oggetti replicati utilizzeranno per impostazione predefinita la classe di archiviazione del bucket di destinazione.
      
      Fai clic su Fine.
  3. Nella sezione Scegli come archiviare i tuoi dati, segui questi passaggi:
    1. Seleziona una classe di archiviazione predefinita per il bucket o Autoclass per la gestione automatica della classe di archiviazione dei dati del bucket.
    2. Per attivare lo spazio dei nomi gerarchico, nella sezione Ottimizza l'archiviazione per workload con uso intensivo dei dati, seleziona Abilita uno spazio dei nomi gerarchico in questo bucket.
      Nota: non puoi abilitare lo spazio dei nomi gerarchico nei bucket esistenti.
  4. Nella sezione Scegli come controllare l'accesso agli oggetti, seleziona se il bucket applica o meno la prevenzione dell'accesso pubblico e seleziona un metodo di controllo dell'accesso per gli oggetti del bucket.
    Nota:non puoi modificare l'impostazione Impedisci l'accesso pubblico se questa impostazione viene applicata a un criterio dell'organizzazione.
  5. Nella sezione Scegli come proteggere i dati degli oggetti, segui questi passaggi:
    - Seleziona una delle opzioni in Protezione dei dati che vuoi impostare per il bucket.
      - Per attivare l'eliminazione temporanea, fai clic sulla casella di controllo Criterio di eliminazione temporanea (per il recupero dei dati) e specifica il numero di giorni per cui vuoi conservare gli oggetti dopo l'eliminazione.
      - Per impostare il controllo delle versioni degli oggetti, seleziona la casella di controllo Controllo delle versioni degli oggetti (per il controllo delle versioni) e specifica il numero massimo di versioni per oggetto e il numero di giorni dopo i quali scadono le versioni non correnti.
      - Per abilitare il criterio di conservazione su oggetti e bucket, seleziona la casella di controllo Conservazione (per la conformità), quindi procedi nel seguente modo:
        
        Per attivare il blocco della conservazione degli oggetti, fai clic sulla casella di controllo Abilita conservazione degli oggetti.
        
        Per attivare Bucket Lock, fai clic sulla casella di controllo Imposta criterio di conservazione del bucket e scegli un'unità di tempo e una durata per il periodo di conservazione.
    - Per scegliere come verranno criptati i dati degli oggetti, espandi la sezione Crittografia dei dati () e seleziona un metodo di crittografia dei dati.
4. Fai clic su Crea.
Imposta le variabili di ambiente locali. Imposta le variabili di ambiente sulla tua macchina locale. Imposta il tuo Google Cloud project-id e il nome del bucket Cloud Storage che utilizzerai per questo tutorial. Fornisci anche il nome e la regione di un cluster Dataproc esistente o nuovo. 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"
```
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
```
Il comando precedente installa la versione immagine cluster predefinita. Puoi utilizzare il flag --image-version per selezionare una versione dell'immagine per il tuo cluster. Ogni versione dell'immagine installa versioni specifiche dei componenti delle librerie Spark e Scala. Se preparare il job Spark wordcount in Java o Scala, farai riferimento alle versioni di Spark e Scala installate sul cluster quando prepari il pacchetto del job.
Copia i dati pubblici nel tuo bucket Cloud Storage. Copia un frammento di testo di Shakespeare pubblico nella cartella input del tuo bucket Cloud Storage:
```
gcloud storage cp gs://pub/shakespeare/rose.txt \
    gs://${BUCKET_NAME}/input/rose.txt
```
Configura un ambiente di sviluppo Java (Apache Maven), Scala (SBT) o Python.
Utilizza Cloud Shell. Cloud Shell include gli strumenti utilizzati in questo tutorial, tra cui Apache Maven, Python e Google Cloud CLI.

Prepara il job Spark di conteggio parole

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

Job Spark in Java utilizzando Apache Maven per creare un pacchetto JAR
Job Spark in Scala utilizzando SBT per creare un pacchetto JAR
Job Spark in Python (PySpark)

Java

Copia il file pom.xml nella macchina locale. Il seguente file pom.xml specifica le dipendenze delle librerie Scala e Spark, a cui viene 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 da 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

Controlla la versione dell'immagine del cluster. Sostituisci i segnaposto version nel file per mostrare le versioni delle librerie Spark e Scala utilizzate dalla versione dell'immagine del cluster. Tieni presente che il numero dell'artefatto spark-core_ è il numero di versione di 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>

Copia il codice WordCount.java elencato di seguito sul tuo computer locale.

Crea un insieme di directory con il percorso src/main/java/dataproc/codelab:
```
mkdir -p src/main/java/dataproc/codelab
```
Copia WordCount.java sulla tua macchina locale in src/main/java/dataproc/codelab:
```
cp WordCount.java src/main/java/dataproc/codelab
```

WordCount.java è un job Spark in Java che legge file di testo da Cloud Storage, esegue un conteggio delle parole e poi scrive i risultati del 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);
  }
}

Crea il pacchetto.
```
mvn clean package
```
Se la build ha esito positivo, viene creato un target/word-count-1.0.jar.

Trasferisci il pacchetto a Cloud Storage.

gcloud storage cp target/word-count-1.0.jar \
    gs://${BUCKET_NAME}/java/word-count-1.0.jar

Scala

Copia il file build.sbt nella macchina locale. Il seguente file build.sbt specifica le dipendenze delle librerie Scala e Spark, a cui viene 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 da 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
Controlla la versione dell'immagine del cluster. Sostituisci i segnaposto version nel file per mostrare le versioni delle librerie Spark e Scala utilizzate dalla versione dell'immagine del cluster.
```
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"
)
```

Copia word-count.scala sulla tua macchina locale. Si tratta di un job Spark in Java che legge file di testo da Cloud Storage, esegue un conteggio delle parole e poi 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)
  }
}

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.

Trasferisci il pacchetto a 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

Copia word-count.py sulla tua macchina locale. Si tratta di un job Spark in Python che utilizza PySpark per leggere i file di testo da Cloud Storage, esegue un conteggio delle parole e poi scrive i risultati del 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/

Visualizzare l'output

Al termine del job, esegui questo comando gcloud CLI per visualizzare l'output del conteggio delle parole.

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

L'output del conteggio delle 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 eliminare le risorse che hai creato in modo che non utilizzino più la quota generando addebiti. Le seguenti sezioni descrivono come eliminare o disattivare queste risorse.

Elimina il progetto

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

Per eliminare il progetto:

Attenzione: l'eliminazione di un progetto ha i seguenti effetti:

L'intero contenuto del progetto viene eliminato. Se hai utilizzato un progetto esistente per, le attività descritte in questo documento, eliminerai anche tutto il lavoro che hai svolto nel progetto.
Gli ID progetto personalizzati non sono più disponibili. Quando hai creato questo progetto, potresti aver creato un ID progetto personalizzato che vuoi utilizzare in futuro. Per conservare gli URL che utilizzano l'ID progetto, ad esempio un URL appspot.com, elimina le risorse selezionate all'interno del progetto anziché eliminare l'intero progetto.

Se intendi esplorare più architetture, tutorial o guide rapide, puoi riutilizzare i progetti ed evitare così di superare i limiti di quota.

Nella console Google Cloud , vai alla pagina Gestisci risorse.
Vai a Gestisci risorse
Nell'elenco dei progetti, seleziona quello che vuoi eliminare, quindi fai clic su Elimina.
Nella finestra di dialogo, digita l'ID del progetto e fai clic su Chiudi per eliminare il progetto.

Elimina il cluster Dataproc

Anziché eliminare il progetto, potresti voler eliminare solo il cluster al suo interno.

Elimina il bucket Cloud Storage

ConsoleGoogle Cloud

Nella console Google Cloud , vai alla pagina Bucket in Cloud Storage.
Vai a Bucket
Fai clic sulla casella di controllo del bucket da eliminare.
Per eliminare il bucket, fai clic su Elimina, quindi segui le istruzioni.

Riga di comando

gcloud storage buckets delete BUCKET_NAME

Passaggi successivi

Vedi Suggerimenti per l'ottimizzazione del job Spark

Utilizzo del connettore Cloud Storage con Apache Spark Mantieni tutto organizzato con le raccolte Salva e classifica i contenuti in base alle tue preferenze.

Obiettivi

Costi

Prima di iniziare

Configura la replica tra bucket

Prepara il job Spark di conteggio parole

Java

Scala

Python

Invia il job

Java

Scala

Python

Visualizzare l'output

Esegui la pulizia

Elimina il progetto

Elimina il cluster Dataproc

Elimina il bucket Cloud Storage

ConsoleGoogle Cloud

Riga di comando

Passaggi successivi

Utilizzo del connettore Cloud Storage con Apache Spark