Managed Service for Apache Spark

1. Panoramica - Serverless per Apache Spark

Managed Service for Apache Spark è un servizio completamente gestito e a scalabilità elevata per l'esecuzione di Apache Spark, Apache Flink, Presto e molti altri strumenti e framework open source. Utilizza Managed Apache Spark per la modernizzazione dei data lake, l'ETL / ELT e la data science sicura su scala planetaria. Managed Apache Spark è anche completamente integrato con diversi servizi Google Cloud, tra cui BigQuery, Cloud Storage, Gemini Enterprise Agent Engine e Knowledge Catalog.

Managed Apache Spark è disponibile in due versioni:

• Managed Apache Spark serverless consente di eseguire job PySpark senza dover configurare l'infrastruttura e la scalabilità automatica. Managed Apache Spark supporta le sessioni e i workload batch di PySpark / notebook.
• I cluster Apache Spark gestiti consentono di gestire un cluster Hadoop YARN per carichi di lavoro Spark basati su YARN, oltre a strumenti open source come Flink e Presto. Puoi personalizzare i tuoi cluster basati su cloud con la scalabilità verticale o orizzontale che preferisci, inclusa la scalabilità automatica.

In questo codelab imparerai diversi modi per utilizzare Dataproc Serverless.

Apache Spark è stato creato in origine per essere eseguito su cluster Hadoop e ha utilizzato YARN come gestore di risorse. La manutenzione dei cluster Hadoop richiede un insieme specifico di competenze e la corretta configurazione di molti parametri diversi sui cluster. Oltre a una serie separata di manopole che Spark richiede all'utente di impostare. Ciò porta a molti scenari in cui gli sviluppatori dedicano più tempo alla configurazione dell'infrastruttura anziché a lavorare sul codice Spark stesso.

Dataproc Serverless elimina la necessità di configurare manualmente i cluster Hadoop o Spark. Dataproc Serverless non viene eseguito su Hadoop e utilizza la propria allocazione dinamica delle risorse per determinare i requisiti di risorse, inclusa la scalabilità automatica. Un piccolo sottoinsieme di proprietà Spark è ancora personalizzabile con Dataproc Serverless, ma nella maggior parte dei casi non è necessario modificarle.

2. Configura

Inizierai configurando l'ambiente e le risorse utilizzate in questo codelab.

Crea un progetto Google Cloud. Puoi utilizzarne uno esistente.

Apri Cloud Shell facendo clic nella barra degli strumenti di Cloud Console.

Cloud Shell fornisce un ambiente shell pronto all'uso che puoi utilizzare per questo codelab.

Cloud Shell imposta il nome del progetto per impostazione predefinita. Verifica eseguendo echo $GOOGLE_CLOUD_PROJECT. Se non vedi l'ID progetto nell'output, impostalo.

export GOOGLE_CLOUD_PROJECT=<your-project-id>

Imposta una regione di Compute Engine per le tue risorse, ad esempio us-central1 o europe-west2.

export REGION=<your-region>

Abilita API

Il codelab utilizza le seguenti API:

• BigQuery
• Dataproc

Abilita le API necessarie. L'operazione richiederà circa un minuto e al termine verrà visualizzato un messaggio di operazione riuscita.

gcloud services enable bigquery.googleapis.com
gcloud services enable dataproc.googleapis.com

Configurare l'accesso alla rete

Dataproc Serverless richiede l'abilitazione dell'accesso privato Google nella regione in cui verranno eseguiti i job Spark, poiché i driver e gli esecutori Spark hanno solo IP privati. Esegui questo comando per abilitarlo nella subnet default.

gcloud compute networks subnets update default \
  --region=${REGION} \
  --enable-private-ip-google-access

Puoi verificare che l'accesso privato Google sia abilitato tramite il seguente comando, che restituirà True o False.

gcloud compute networks subnets describe default \
  --region=${REGION} \
  --format="get(privateIpGoogleAccess)"

Crea un bucket di archiviazione

Crea un bucket di archiviazione che verrà utilizzato per archiviare gli asset creati in questo codelab.

Scegli un nome per il bucket. I nomi dei bucket devono essere univoci a livello globale per tutti gli utenti.

export BUCKET=<your-bucket-name>

Crea il bucket nella regione in cui intendi eseguire i job Spark.

gsutil mb -l ${REGION} gs://${BUCKET}

Puoi vedere che il bucket è disponibile nella console di Cloud Storage. Puoi anche eseguire gsutil ls per visualizzare il bucket.

Crea un server di cronologia permanente

L'interfaccia utente di Spark fornisce un ricco set di strumenti di debug e approfondimenti sui job Spark. Per visualizzare l'interfaccia utente di Spark per i job Dataproc Serverless completati, devi creare un cluster Dataproc a nodo singolo da utilizzare come server di cronologia permanente.

Imposta un nome per il server di cronologia permanente.

PHS_CLUSTER_NAME=my-phs

Esegui questo comando.

gcloud dataproc clusters create ${PHS_CLUSTER_NAME} \
    --region=${REGION} \
    --single-node \
    --enable-component-gateway \
    --properties=spark:spark.history.fs.logDirectory=gs://${BUCKET}/phs/*/spark-job-history

L'interfaccia utente di Spark e il server di cronologia permanente verranno esplorati in modo più dettagliato più avanti nel codelab.

3. Esegui job Serverless Spark con Dataproc Batches

In questo esempio, utilizzerai un set di dati del set di dati pubblico Citi Bike Trips di New York (NYC). NYC Citi Bikes è un sistema di bike sharing a pagamento all'interno di NYC. Esegui alcune semplici trasformazioni e stampa i primi dieci ID delle stazioni Citi Bike più utilizzati. Questo esempio utilizza anche in modo significativo spark-bigquery-connector open source per leggere e scrivere dati senza problemi tra Spark e BigQuery.

Clona il seguente repository GitHub e cd nella directory contenente il file citibike.py.

git clone https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos/data-analytics/next-2022-workshop/dataproc-serverless

citibike.py

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import BooleanType

if len(sys.argv) == 1:
    print("Please provide a GCS bucket name.")

bucket = sys.argv[1]
table = "bigquery-public-data:new_york_citibike.citibike_trips"

spark = SparkSession.builder \
          .appName("pyspark-example") \
          .config("spark.jars","gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar") \
          .getOrCreate()

df = spark.read.format("bigquery").load(table)

top_ten = df.filter(col("start_station_id") \
            .isNotNull()) \
            .groupBy("start_station_id") \
            .count() \
            .orderBy("count", ascending=False) \
            .limit(10) \
            .cache()

top_ten.show()

top_ten.write.option("header", True).csv(f"gs://{bucket}/citibikes_top_ten_start_station_ids")

Invia il job a Serverless Spark utilizzando Cloud SDK, disponibile in Cloud Shell per impostazione predefinita. Esegui il comando seguente nella shell, che utilizza Cloud SDK e l'API Dataproc Batches per inviare job Spark serverless.

gcloud dataproc batches submit pyspark citibike.py \
  --batch=citibike-job \
  --region=${REGION} \
  --deps-bucket=gs://${BUCKET} \
  --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar \
--history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER_NAME} \
  -- ${BUCKET}

Per analizzare la situazione nel dettaglio:

• gcloud dataproc batches submit fa riferimento all'API Dataproc Batches.
• pyspark indica che stai inviando un job PySpark.
• --batch è il nome del job. Se non viene fornito, verrà utilizzato un UUID generato casualmente.
• --region=${REGION} è la regione geografica in cui verrà elaborato il job.
• --deps-bucket=${BUCKET} è la posizione in cui viene caricato il file Python locale prima dell'esecuzione nell'ambiente serverless.
• --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar include il file JAR per spark-bigquery-connector nell'ambiente di runtime Spark.
• --history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER} è il nome completo del server di cronologia permanente. È qui che vengono archiviati i dati sugli eventi di Spark (separati dall'output della console) e visualizzabili dall'interfaccia utente di Spark.
• Il -- finale indica che tutto ciò che segue saranno argomenti di runtime per il programma. In questo caso, invii il nome del bucket, come richiesto dal job.

Quando il batch viene inviato, viene visualizzato il seguente output.

Batch [citibike-job] submitted.

Dopo un paio di minuti, vedrai il seguente output insieme ai metadati del job.

+----------------+------+
|start_station_id| count|
+----------------+------+
|             519|551078|
|             497|423334|
|             435|403795|
|             426|384116|
|             293|372255|
|             402|367194|
|             285|344546|
|             490|330378|
|             151|318700|
|             477|311403|
+----------------+------+

Batch [citibike-job] finished.

Nella prossima sezione, imparerai a individuare i log per questo job.

Altre funzionalità

Con Spark Serverless, hai a disposizione opzioni aggiuntive per l'esecuzione dei job.

• Puoi creare un'immagine Docker personalizzata su cui viene eseguito il job. Questo è un ottimo modo per includere dipendenze aggiuntive, tra cui le librerie Python e R.
• Puoi connettere un'istanza Dataproc Metastore al tuo job per accedere ai metadati Hive.
• Per un maggiore controllo, Dataproc Serverless supporta la configurazione di un piccolo insieme di proprietà Spark.

4. Metriche e osservabilità di Dataproc

La console Batch Dataproc elenca tutti i tuoi job Dataproc Serverless. Nella console vedrai ID batch, Posizione, Stato, Ora di creazione, Tempo trascorso e Tipo di ogni job. Fai clic sull'ID batch del job per visualizzare ulteriori informazioni.

In questa pagina vedrai informazioni come Monitoraggio, che mostra quanti esecutori Spark batch sono stati utilizzati nel tempo dal tuo job (indicando il livello di scalabilità automatica).

Nella scheda Dettagli vedrai altri metadati sul job, inclusi gli argomenti e i parametri inviati con il job.

Puoi anche accedere a tutti i log da questa pagina. Quando vengono eseguiti i job Dataproc Serverless, vengono generati tre diversi set di log:

• Livello di servizio
• Output console
• Registrazione eventi Spark

A livello di servizio, include i log generati dal servizio Dataproc Serverless. Questi includono elementi come Dataproc Serverless che richiede CPU aggiuntive per la scalabilità automatica. Puoi visualizzarli facendo clic su Visualizza log, che aprirà Cloud Logging.

L'output della console può essere visualizzato in Output.Si tratta dell'output generato dal job, inclusi i metadati stampati da Spark all'inizio di un job o le istruzioni di stampa incorporate nel job.

La registrazione degli eventi Spark è accessibile dall'interfaccia utente di Spark. Poiché hai fornito al tuo job Spark un server di cronologia permanente, puoi accedere all'interfaccia utente di Spark facendo clic su Visualizza server di cronologia Spark, che contiene informazioni sui job Spark eseguiti in precedenza. Puoi scoprire di più sull'interfaccia utente di Spark nella documentazione di Spark ufficiale.

5. Modelli Dataproc: BQ -> GCS

I modelli Dataproc sono strumenti open source che contribuiscono a semplificare ulteriormente le attività di elaborazione dei dati nel cloud. Fungono da wrapper per Dataproc Serverless e includono modelli per molte attività di importazione ed esportazione dei dati, tra cui:

• BigQuerytoGCS e GCStoBigQuery
• GCStoBigTable
• GCStoJDBC e JDBCtoGCS
• HivetoBigQuery
• MongotoGCS e GCStoMongo

L'elenco completo è disponibile nel file README.

In questa sezione utilizzerai i modelli Dataproc per esportare i dati da BigQuery a GCS.

Clona il repository

Clona il repository e passa alla cartella python.

git clone https://github.com/GoogleCloudPlatform/dataproc-templates.git
cd dataproc-templates/python

configura l'ambiente

Ora imposterai le variabili di ambiente. I modelli Dataproc utilizzano la variabile di ambiente GCP_PROJECT per l'ID progetto, quindi impostala su GOOGLE_CLOUD_PROJECT.

export GCP_PROJECT=${GOOGLE_CLOUD_PROJECT}

La tua regione deve essere impostata nell'ambiente precedente. In caso contrario, impostalo qui.

export REGION=<region>

I modelli Dataproc utilizzano spark-bigquery-connector per l'elaborazione dei job BigQuery e richiedono che l'URI sia incluso in una variabile di ambiente JARS. Imposta la variabile JARS.

export JARS="gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar"

Configurare i parametri del modello

Imposta il nome di un bucket di staging da utilizzare per il servizio.

export GCS_STAGING_LOCATION=gs://${BUCKET}

Successivamente, imposterai alcune variabili specifiche del job. Per la tabella di input, farai di nuovo riferimento al set di dati BigQuery NYC Citibike.

BIGQUERY_GCS_INPUT_TABLE=bigquery-public-data.new_york_citibike.citibike_trips

Puoi scegliere csv, parquet, avro o json. Per questo codelab, scegli CSV. Nella sezione successiva vedrai come utilizzare i modelli Dataproc per convertire i tipi di file.

BIGQUERY_GCS_OUTPUT_FORMAT=csv

Imposta la modalità di output su overwrite. Puoi scegliere tra overwrite, append, ignore o errorifexists.

BIGQUERY_GCS_OUTPUT_MODE=overwrite

Imposta la posizione di output GCS in modo che sia un percorso nel bucket.

BIGQUERY_GCS_OUTPUT_LOCATION=gs://${BUCKET}/BQtoGCS

Esegui il modello

Esegui il modello BIGQUERYTOGCS specificandolo di seguito e fornendo i parametri di input che hai impostato.

./bin/start.sh \
-- --template=BIGQUERYTOGCS \
        --bigquery.gcs.input.table=${BIGQUERY_GCS_INPUT_TABLE} \
        --bigquery.gcs.output.format=${BIGQUERY_GCS_OUTPUT_FORMAT} \
        --bigquery.gcs.output.mode=${BIGQUERY_GCS_OUTPUT_MODE} \
        --bigquery.gcs.output.location=${BIGQUERY_GCS_OUTPUT_LOCATION}

L'output sarà piuttosto rumoroso, ma dopo circa un minuto vedrai quanto segue.

Batch [5766411d6c78444cb5e80f305308d8f8] submitted.
...
Batch [5766411d6c78444cb5e80f305308d8f8] finished.

Per verificare che i file siano stati generati, esegui il seguente comando.

gsutil ls ${BIGQUERY_GCS_OUTPUT_LOCATION}

Per impostazione predefinita, Spark scrive in più file, a seconda della quantità di dati. In questo caso, vedrai circa 30 file generati. I nomi dei file di output di Spark sono formattati con part, seguito da un numero di cinque cifre (che indica il numero di parte) e da una stringa hash. Per grandi quantità di dati, Spark in genere scrive in più file. Un esempio di nome file è part-00000-cbf69737-867d-41cc-8a33-6521a725f7a0-c000.csv.

6. Modelli Dataproc: da CSV a Parquet

Ora utilizzerai i modelli Dataproc per convertire i dati in GCS da un tipo di file a un altro utilizzando GCSTOGCS. Questo modello utilizza SparkSQL e offre la possibilità di inviare anche una query SparkSQL da elaborare durante la trasformazione per un'ulteriore elaborazione.

Conferma le variabili di ambiente

Verifica che GCP_PROJECT, REGION e GCS_STAGING_BUCKET siano impostati nella sezione precedente.

echo ${GCP_PROJECT}
echo ${REGION}
echo ${GCS_STAGING_LOCATION}

Imposta parametri del modello

Ora imposterai i parametri di configurazione per GCStoGCS. Inizia con la posizione dei file di input. Tieni presente che si tratta di una directory e non di un file specifico, poiché verranno elaborati tutti i file nella directory. Imposta questo valore su BIGQUERY_GCS_OUTPUT_LOCATION.

GCS_TO_GCS_INPUT_LOCATION=${BIGQUERY_GCS_OUTPUT_LOCATION}

Imposta il formato del file di input.

GCS_TO_GCS_INPUT_FORMAT=csv

Imposta il formato di output desiderato. Puoi scegliere parquet, json, avro o csv.

GCS_TO_GCS_OUTPUT_FORMAT=parquet

Imposta la modalità di output su overwrite. Puoi scegliere tra overwrite, append, ignore o errorifexists.

GCS_TO_GCS_OUTPUT_MODE=overwrite

Imposta la posizione di output.

GCS_TO_GCS_OUTPUT_LOCATION=gs://${BUCKET}/GCStoGCS

Esegui il modello

Esegui il modello GCStoGCS.

./bin/start.sh \
-- --template=GCSTOGCS \
        --gcs.to.gcs.input.location=${GCS_TO_GCS_INPUT_LOCATION} \
        --gcs.to.gcs.input.format=${GCS_TO_GCS_INPUT_FORMAT} \
        --gcs.to.gcs.output.format=${GCS_TO_GCS_OUTPUT_FORMAT} \
        --gcs.to.gcs.output.mode=${GCS_TO_GCS_OUTPUT_MODE} \
        --gcs.to.gcs.output.location=${GCS_TO_GCS_OUTPUT_LOCATION}

L'output sarà piuttosto rumoroso, ma dopo circa un minuto dovresti visualizzare un messaggio di operazione riuscita simile al seguente.

Batch [c198787ba8e94abc87e2a0778c05ec8a] submitted.
...
Batch [c198787ba8e94abc87e2a0778c05ec8a] finished.

Per verificare che i file siano stati generati, esegui il seguente comando.

gsutil ls ${GCS_TO_GCS_OUTPUT_LOCATION}

Con questo modello, hai anche la possibilità di fornire query SparkSQL passando gcs.to.gcs.temp.view.name e gcs.to.gcs.sql.query al modello, consentendo l'esecuzione di una query SparkSQL sui dati prima della scrittura in GCS.

7. Libera spazio

Per evitare che al tuo account GCP vengano addebitati costi inutili dopo il completamento di questo codelab:

1. Elimina il bucket Cloud Storage per l'ambiente che hai creato.

gsutil rm -r gs://${BUCKET}

2. Elimina il cluster Dataproc utilizzato per il server di cronologia permanente.

gcloud dataproc clusters delete ${PHS_CLUSTER_NAME} \
  --region=${REGION}

3. Elimina i job Dataproc Serverless. Vai alla console Batches, fai clic sulla casella accanto a ogni job che vuoi eliminare e fai clic su ELIMINA.

Se hai creato un progetto solo per questo codelab, puoi anche eliminarlo facoltativamente:

1. Nella console di GCP, vai alla pagina Progetti.
2. Nell'elenco dei progetti, seleziona quello da eliminare e fai clic su Elimina.
3. Nella casella, digita l'ID progetto, quindi fai clic su Chiudi 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 questo tutorial, quando lo eliminerai, eliminerai anche tutto il lavoro effettuato 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.

8. Passaggi successivi

Le seguenti risorse forniscono ulteriori modi per sfruttare Serverless Spark:

• Scopri come orchestrare i flussi di lavoro Dataproc Serverless utilizzando Cloud Composer.
• Scopri come integrare Dataproc Serverless con le pipeline Kubeflow.