Google PDE: Spark, Dataproc e Elaborazione Distribuita dei Dati — Guida allo studio

By Laurynas Brazys · 21 August 2026 · #google #PDE #certification #exam-prep #study-guide

Fa parte della Google Professional Data Engineer — Guida allo studio. Esercitati con risposte verificate nel centro esami Google, oppure fai test cronometrati su ExamRoll.io.

Panoramica

Apache Spark su Google Cloud Dataproc fornisce una piattaforma gestita ed elastica per l’elaborazione distribuita dei dati. È possibile scegliere tra cluster Dataproc a lunga esecuzione o effimeri e Dataproc Serverless per Spark, a seconda delle esigenze di controllo, della variabilità del runtime e dell’overhead di gestione. Spark offre astrazioni resilienti (RDD), API relazionali (DataFrame e Spark SQL) e un motore di esecuzione DAG fault-tolerant ottimizzato per ETL iterativi e batch su larga scala. Su Google Cloud, Cloud Storage sostituisce HDFS per uno storage durevole e a basso costo; il connettore BigQuery consente un offload analitico diretto; e Dataproc Metastore centralizza la gestione degli schemi. Le soluzioni efficaci allineano i cicli di vita dello storage e del calcolo, ottimizzano Spark per il carico di lavoro, strumentano l’osservabilità e applicano la sicurezza con il principio del minimo privilegio e l’isolamento di rete.

Architettura di Dataproc: Cluster, Serverless, Storage e Metastore

• Tipi di cluster e ruoli dei nodi
  • I nodi primari (master) ospitano YARN, l’HDFS NameNode (se utilizzato), le interfacce utente del driver Spark; la modalità HA utilizza più nodi primari.
  • I nodi worker eseguono gli executor e gli HDFS DataNode (se utilizzati).
  • I worker secondari/ausiliari sono tipicamente preemptible/spot per una capacità elastica a basso costo, senza ruoli HDFS.
  • Le immagini raggruppano versioni del sistema operativo e dei componenti (ad esempio, 2.1-debian11, 2.2-ubuntu20); fissare le versioni delle immagini per controllare la compatibilità di Spark/Hadoop ed eseguire gli aggiornamenti in modo deliberato.
  • Component Gateway pubblica le interfacce utente (Spark History Server, YARN RM) in modo sicuro tramite HTTPS.
• Dataproc Serverless per Spark
  • Nessun provisioning di cluster, autoscaling automatico e fatturazione al secondo per executor e driver. Ideale per job sporadici o con picchi di carico, o quando si vuole ridurre al minimo l’overhead operativo.
  • Compromessi: meno controlli di basso livello rispetto ai cluster; la latenza di avvio dei job può essere superiore a quella dei cluster “caldi”; utilizzare le metriche e i log degli eventi serverless per il troubleshooting.
• Autoscaling
  • Le policy di autoscaling del cluster aggiungono/rimuovono worker in base alle metriche di YARN/Spark e ai periodi di cooldown, ottimizzando separatamente i gruppi di worker primari e secondari.
  • L’autoscaling serverless è gestito dal servizio; progettare le applicazioni per essere parallelizzabili per partizione ed evitare colli di bottiglia serializzati per ottenere la migliore scalabilità.
• Storage e connettori
  • Preferire Google Cloud Storage (GCS) come sistema di riferimento; disaccoppia il calcolo dallo storage, riduce i costi dei dischi permanenti e sopravvive ai cicli di vita dei cluster.
  • Il connettore GCS (gs://) si integra con Hadoop/Spark. Le scritture su object store utilizzano protocolli di commit; impostare l’algoritmo FileOutputCommitter v2 per ridurre l’overhead di ridenominazione e accelerare il commit dei job su GCS:

    --conf mapreduce.fileoutputcommitter.algorithm.version=2
    ```

  - Utilizzare Parquet/ORC con column pruning e predicate pushdown. Gestire i file di piccole dimensioni tramite compattazione per raggiungere una dimensione target di 128–512 MiB per file per una scansione efficiente.
- Hive metastore
  - Centralizzare schemi e metadati delle tabelle in Dataproc Metastore (un Apache Hive Metastore gestito) o in un metastore basato su Cloud SQL per condividere i cataloghi tra più cluster.
  - Utilizzare tabelle esterne che puntano a GCS per la durabilità; partizionare per data/ora per limitare i costi di scansione.
- Job, inizializzazione e workflow
  - Inviare job spark, pyspark, spark-sql o hadoop. Le azioni di inizializzazione installano librerie o agenti aggiuntivi alla creazione del cluster (ad esempio, connettori, librerie Python).
  - I modelli di workflow parametrizzano pipeline multi-step; possono creare cluster effimeri per ogni workflow e poi eliminarli. Ciò migliora l'isolamento e riduce i costi di inattività.
  - I cluster effimeri sono consigliati per ETL batch; i dati e il metastore risiedono all'esterno del cluster (GCS, Dataproc Metastore, BigQuery).
- Integrazione con BigQuery
  - Il connettore Spark BigQuery legge/scrive direttamente su BigQuery; considerare la BigQuery Storage Read API per il throughput e la Write API per inserimenti in streaming a bassa latenza e con semantica "exactly-once".
  - Per la manutenzione delle tabelle, eseguire operazioni MERGE o di sovrascrittura delle partizioni a valle in BigQuery per finalizzare i caricamenti in modo atomico.
### Modello Spark, Ottimizzazione delle Prestazioni e Affidabilità

- API ed esecuzione
  - RDD: di basso livello, immutabili, type-safe in Scala/Java; si controllano il partizionamento e la persistenza.
  - DataFrame/Dataset: relazionali, ottimizzati con Catalyst; preferire questi per l'ETL grazie all'ottimizzazione delle query e alla generazione di codice.
  - Le trasformazioni sono lazy (map, filter, join); le azioni ne attivano l'esecuzione (count, collect, save). Spark costruisce un DAG di stage separati da shuffle; i task vengono eseguiti per ogni partizione.
- Partizionamento e shuffle
  - Partizionamento in input: un numero di partizioni sufficiente a utilizzare tutti i core; iniziare con un valore pari a 2-4 volte il numero totale di core degli executor. Controllare tramite `spark.default.parallelism` (per gli RDD) e le opzioni del reader (per i DataFrame).
  - Partizioni di shuffle: il valore predefinito di 200 è spesso insufficiente o eccessivo. Ottimizzare:

--conf spark.sql.shuffle.partitions= {total_executor_cores * 2 to 3}
```

• Puntare a circa 100-256 MiB per partizione dopo trasformazioni wide; partizioni troppo piccole causano overhead dello scheduler, mentre partizioni troppo grandi rischiano di causare OOM nell’executor.
• Lo shuffle è il costo dominante per le operazioni di join, groupBy e orderBy. Assicurare memoria e disco adeguati per l’executor; considerare SSD locali per cluster con shuffle intensivo.
• Skew e strategia di join
  • Rilevare lo skew (tempi di esecuzione dei task con long-tail, partizioni di grandi dimensioni). Mitigazioni:
    • Eseguire il broadcast di tabelle piccole per evitare gli shuffle:

      --conf spark.sql.autoBroadcastJoinThreshold=64m
      ```

    - Aggiungere un "salt" alle chiavi per le partizioni "hot"; applicare una pre-aggregazione lato map; filtrare i dati il prima possibile.
    - Abilitare l'Adaptive Query Execution (AQE) per unire le partizioni post-shuffle e gestire gli skewed join:

  --conf spark.sql.adaptive.enabled=true
  ```

• Caching, checkpointing e lineage
  • Mettere in cache i DataFrame intermedi “hot” con parsimonia quando vengono riutilizzati; preferire MEMORY_AND_DISK per evitare OOM.
  • Eseguire il checkpoint di lineage lunghi su GCS o HDFS per limitare il ricalcolo in caso di fallimenti.
• Executor e allocazione dinamica
  • Dimensionare correttamente gli executor per bilanciare parallelismo e overhead del GC:
    • Core per executor: 2-5 per task bilanciati tra I/O e CPU; un numero inferiore di core riduce le pause del GC.
    • Overhead di memoria: impostare spark.yarn.executor.memoryOverhead per shuffle di tipo wide.
    • Abilitare l’allocazione dinamica con un external shuffle service sui cluster per scalare gli executor in base al carico di lavoro:

      --conf spark.dynamicAllocation.enabled=true
      --conf spark.shuffle.service.enabled=true
      --conf spark.dynamicAllocation.minExecutors=0
      --conf spark.dynamicAllocation.maxExecutors=200
      ```

- Pattern di tolleranza ai guasti per ETL batch
  - Scritture idempotenti: scrivere su un percorso temporaneo/di staging, quindi promuovere atomicamente con un commit a livello di directory; per BigQuery, scrivere su una tabella di staging ed eseguire un `MERGE`:

MERGE target t USING staging s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT (...)
```

• Elaborazione incrementale: usare filtri basati su watermark sulle partizioni ingestion_date; mantenere un manifest dei dati elaborati in GCS per evitare la rielaborazione.
• Gestione dei dead-letter: in caso di errori di parsing/validazione, deviare i record errati verso un percorso/tabella di quarantena con informazioni diagnostiche. Per un’applicazione rigorosa dello schema e DLQ integrate, considerare Dataflow; con Spark, implementare un blocco try/catch per ogni record e un sink separato.

Sicurezza, Osservabilità e Costi

• Identità e accesso
  • Esegui cluster e job con service account dedicati secondo il principio del privilegio minimo (least-privilege) di IAM. Concedi solo i ruoli necessari, ad esempio:
    • roles/dataproc.worker ai service account delle istanze
    • roles/storage.objectViewer o objectAdmin per i percorsi di I/O di GCS
    • roles/bigquery.dataEditor sui set di dati di destinazione
  • Per Dataproc Serverless, utilizza service account per-job per definire l’ambito dell’accesso.
• Isolamento della rete e crittografia
  • Utilizza cluster con IP privato in una subnet VPC, limita l’accesso alle UI del master tramite firewall e abilita l’Accesso Privato Google per GCS/BigQuery senza egress pubblico.
  • Posiziona i cluster in progetti Shared VPC per un controllo centralizzato. Opzionalmente, abilita Kerberos su Dataproc per l’autenticazione all’interno del cluster.
  • Crittografa i dati a riposo (at rest) con CMEK: configura CMEK su bucket GCS, Persistent Disk, Dataproc Metastore e BigQuery; utilizza TLS per i dati in transito (in transit) per impostazione predefinita.
• Logging, cronologia e metriche
  • Abilita i log degli eventi di Spark su GCS e distribuisci l’History Server:

    --conf spark.eventLog.enabled=true
    --conf spark.eventLog.dir=gs://bucket/spark-events/
    ```

  - Dataproc invia in streaming i log del driver e di YARN a Cloud Logging; esportali verso i sink per la conservazione/analisi forense.
  - Monitora con le metriche di Cloud Monitoring: container YARN in attesa, CPU, memoria, stato di HDFS (se utilizzato), throughput di GCS. Imposta avvisi (alert) per tentativi di stage prolungati, perdita di executor e picchi di esecuzione speculativa.
  - Analisi dei fallimenti: le cause comuni includono straggler indotti da skew, OOM (Out of Memory) degli executor durante lo shuffle, fallimenti nel commit sull'object-store e perdita di nodi preemptible/spot. Aumenta il numero di tentativi (retry) con giudizio; tentativi eccessivi possono amplificare costi e ritardi.
- Ottimizzazione dei costi
  - Utilizza cluster effimeri o Dataproc Serverless per evitare costi di inattività; mantieni i dati in GCS per ridurre al minimo l'uso di dischi permanenti.
  - Aggiungi worker secondari preemptible/spot per assorbire i picchi di domanda; progetta per la rielaborazione, poiché i task sui nodi persi vengono ritentati. Non posizionare i nodi master su nodi preemptible.
  - Dimensiona correttamente i tipi di macchina (right-sizing) e usa l'autoscaling per ridurre la capacità quando le code sono vuote. Prediligi Parquet/ORC con partition pruning per ridurre i costi di scansione e l'uso della CPU.
  - Evita file di piccole dimensioni compattando gli output; un numero minore di file più grandi riduce l'overhead dei metadati e il tempo di esecuzione del job.
  - Per job brevi e periodici (ad esempio, un ETL Spark settimanale di 30 minuti), i worker preemptible o la modalità serverless offrono spesso il miglior profilo di costo.

#### Scenario Pratico

Acme Retail sta migrando un cluster Hadoop on-premise da 30 nodi che esegue ETL notturni con Spark e Hive per alimentare sistemi di analytics a valle. L'obiettivo è riutilizzare i job esistenti con modifiche minime, evitare la gestione a tempo pieno dei cluster, persistere i dati oltre il ciclo di vita del cluster e ridurre i costi di storage.

Approccio:
1) Spostare dati e metadati su servizi gestiti
   - Archivia tutti i dati grezzi e curati in Cloud Storage utilizzando Parquet con partizionamento (ad esempio, dt=AAAA-MM-GG).
   - Motivazione: GCS è durevole, a basso costo e disaccoppia il calcolo (compute) dallo storage, consentendo a cluster effimeri e job serverless di funzionare senza dischi permanenti. Il formato Parquet partizionato abilita il predicate pushdown e scansioni efficienti.

2) Centralizzare il catalogo con Dataproc Metastore
   - Migra l'Hive metastore su Dataproc Metastore. Crea tabelle Hive esterne che fanno riferimento ai percorsi GCS e mantieni la logica di schema/partizionamento esistente.
   - Motivazione: Un metastore gestito consente a più cluster effimeri e job serverless di condividere le definizioni delle tabelle senza dover eseguire un'istanza HA di MySQL/PostgreSQL.

3) Utilizzare cluster Dataproc effimeri per l'ETL batch e workflow template per l'orchestrazione
   - Definisci un workflow template che crea un cluster con l'immagine richiesta (ad esempio, 2.1-debian11), esegue i job Spark (spark-sql e pyspark) ed elimina il cluster al termine. Aggiungi azioni di inizializzazione per installare eventuali librerie personalizzate.
   - Motivazione: I cluster effimeri eliminano i costi di inattività e isolano le dipendenze dei job. I workflow template offrono ripetibilità e parametrizzazione (date, percorsi di input).

4) Abilitare l'autoscaling e i worker preemptible
   - Associa una policy di autoscaling con un piccolo gruppo di worker principali (core) e un pool più ampio di worker secondari preemptible; ottimizza i periodi di cooldown per ridurre rapidamente la capacità dopo l'esecuzione.
   - Motivazione: I worker principali mantengono la stabilità del cluster; i worker preemptible assorbono gli shuffle e le trasformazioni ampie (wide transformations) a un costo inferiore. I tentativi (retry) di Spark/YARN gestiscono i task persi a causa della preemption.

5) Integrare con BigQuery tramite lo Spark BigQuery connector
   - Per i caricamenti di dimensioni/fatti, scrivi i risultati di Spark in tabelle di staging di BigQuery, quindi esegui istruzioni MERGE per aggiornare le tabelle di destinazione in modo atomico. Dove la sovrascrittura diretta è sicura, scrivi su tabelle partizionate utilizzando la modalità di sovrascrittura della partizione (partition overwrite mode).
   - Motivazione: BigQuery serve analytics e BI su larga scala; la combinazione staging+MERGE produce upsert di tipo transazionale da job Spark batch, riducendo l'incoerenza a valle.

6) Ottimizzare Spark per prestazioni e affidabilità
   - Imposta le partizioni di shuffle in relazione ai core degli executor e abilita AQE:

 --conf spark.sql.shuffle.partitions=600
 --conf spark.sql.adaptive.enabled=true
 ```

• Utilizza broadcast join per le dimensioni piccole e crea checkpoint su GCS per i lineage lunghi per garantire la stabilità.
• Motivazione: Un partizionamento corretto riduce lo skew e l’overhead dello scheduler; AQE si adatta ai profili dei dati in fase di esecuzione; il checkpointing limita la rielaborazione dopo i fallimenti.

7. Rafforzare la sicurezza e la rete

   • Esegui i cluster con service account dedicati, concedendo solo i ruoli necessari per i percorsi GCS, il metastore e i set di dati BigQuery. Crea cluster con IP privato in una subnet ristretta con Accesso Privato Google e limita l’accesso alla UI tramite regole di firewall.
   • Motivazione: Il principio del privilegio minimo e l’isolamento della rete riducono la superficie di attacco; l’egress privato del piano di controllo evita l’esposizione pubblica.

8. Strumentare logging, cronologia e avvisi

   • Abilita i log degli eventi di Spark su GCS e distribuisci l’History Server; instrada i log del driver/YARN a Cloud Logging con una policy di conservazione. Aggiungi avvisi di Monitoring per container in attesa per lungo tempo, fallimenti ripetuti dei task o durata eccessiva del job.
   • Motivazione: I log centralizzati supportano l’analisi delle cause principali (root-cause analysis); gli avvisi proattivi rilevano precocemente skew, OOM o I/O degradato.

9. Modernizzare selettivamente con Dataproc Serverless per carichi ad hoc e picchi elastici

   • Sposta i carichi di lavoro Spark SQL sporadici o esplorativi su Dataproc Serverless; mantieni le pipeline notturne su cluster effimeri finché non saranno completamente convalidate su serverless.
   • Motivazione: La modalità serverless elimina le operazioni di gestione del cluster e scala automaticamente, ideale per carichi imprevedibili; i workflow esistenti continuano a funzionare con modifiche minime al codice.

10. Convalidare i committer per l’object-store e la gestione dei file di piccole dimensioni

    • Imposta l’algoritmo v2 di FileOutputCommitter e compatta gli output in file da 256–512 MiB tramite repartition/coalesce prima delle scritture.
    • Motivazione: Gli object store non dispongono di una ridenominazione atomica; i committer ottimizzati riducono l’overhead di copia/ridenominazione. La compattazione mitiga il problema dei file di piccole dimensioni, migliorando prestazioni e costi.

Questo design riutilizza i job Spark e Hive esistenti con un refactoring minimo, garantisce la durabilità dei dati in GCS, centralizza gli schemi, contiene il raggio d’azione di un’eventuale violazione della sicurezza (security blast radius), fornisce un’osservabilità robusta e ottimizza i costi attraverso cluster effimeri, autoscaling, capacità preemptible e l’uso mirato dell’esecuzione serverless.

← Messaggistica · Tutti i domini · Ingestione →

Esercitati su queste domande → · Pratica cronometrata su ExamRoll.io →

Sfoglia Google →