Google PDE: Spark, Dataproc and Distributed Data Processing — Study Guide

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

Part of the Google Professional Data Engineer — Study Guide. Practice with verified answers in the Google exam hub, or take timed practice tests on ExamRoll.io.

Overview

Apache Spark on Google Cloud Dataproc provides a managed, elastic platform for distributed data processing. You can choose between long-running or ephemeral Dataproc clusters and Dataproc Serverless for Spark, depending on control needs, runtime variability, and management overhead. Spark offers resilient abstractions (RDDs), relational APIs (DataFrames and Spark SQL), and a fault-tolerant DAG execution engine optimized for iterative and batch ETL at scale. On Google Cloud, Cloud Storage replaces HDFS for durable, low-cost storage; the BigQuery connector enables direct analytical offload; and Dataproc Metastore centralizes schema management. Effective solutions align storage and compute lifecycles, tune Spark to the workload, instrument observability, and apply security with least privilege and network isolation.

Dataproc Architecture: Clusters, Serverless, Storage, and Metastore

Cluster types and node roles
- Primary (master) nodes host YARN, HDFS NameNode (if used), Spark driver UIs; HA mode uses multiple primaries.
- Worker nodes run executors and HDFS DataNodes (if used).
- Secondary/auxiliary workers are typically preemptible/spot for elastic, lower-cost capacity without HDFS roles.
- Images bundle OS and component versions (for example, 2.1-debian11, 2.2-ubuntu20); pin image versions to control Spark/Hadoop compatibility and upgrade deliberately.
- Component Gateway publishes UIs (Spark History Server, YARN RM) securely via HTTPS.
Dataproc Serverless for Spark
- No cluster provisioning, automatic autoscaling, and per-second billing for executors and drivers. Ideal for sporadic or bursty jobs, or when minimizing ops overhead.
- Trade-offs: fewer low-level knobs than clusters; job start latency may be higher than warm clusters; use serverless metrics and event logs for troubleshooting.
Autoscaling
- Cluster autoscaling policies add/remove workers based on YARN/Spark metrics and cooldowns, separately tuning primary and secondary worker groups.
- Serverless autoscaling is managed by the service; design to be partition-parallel and avoid serialized bottlenecks for best scaling.
Storage and connectors
- Prefer Google Cloud Storage (GCS) as the system-of-record; it decouples compute from storage, reduces persistent-disk cost, and survives cluster lifecycles.
- The GCS connector (gs://) integrates with Hadoop/Spark. Writes to object stores use commit protocols; set FileOutputCommitter algorithm v2 to reduce rename overhead and speed up job commits on GCS:
```
--conf mapreduce.fileoutputcommitter.algorithm.version=2
```
- Use Parquet/ORC with column pruning and predicate pushdown. Manage small files via compaction to target 128–512 MiB per file for efficient scan.
Hive metastore
- Centralize schemas and table metadata in Dataproc Metastore (managed Apache Hive Metastore) or Cloud SQL–backed metastore to share catalogs across clusters.
- Use external tables pointing at GCS for durability; partition by date/hour to bound scan cost.
Jobs, initialization, and workflows
- Submit spark, pyspark, spark-sql, or hadoop jobs. Initialization actions install additional libraries or agents at cluster creation (for example, connectors, Python libs).
- Workflow templates parameterize multi-step pipelines; they can create ephemeral clusters per workflow, then tear them down. This improves isolation and cuts idle cost.
- Ephemeral clusters are recommended for batch ETL; data and metastore live outside the cluster (GCS, Dataproc Metastore, BigQuery).
BigQuery integration
- The Spark BigQuery connector reads/writes BigQuery directly; consider BigQuery Storage Read API for throughput and the Write API for lower-latency, exactly-once streaming inserts.
- For table maintenance, perform downstream MERGE/partition overwrites in BigQuery to finalize loads atomically.

Spark Model, Performance Tuning, and Reliability

APIs and execution
- RDDs: low-level, immutable, type-safe in Scala/Java; you control partitioning and persistence.
- DataFrames/Datasets: relational, Catalyst-optimized; prefer these for ETL due to query optimization and code generation.
- Transformations are lazy (map, filter, join); actions trigger execution (count, collect, save). Spark builds a DAG of stages split by shuffles; tasks run per partition.
Partitioning and shuffle
- Input partitioning: enough partitions to utilize all cores; start with 2–4x total executor cores. Control via spark.default.parallelism (for RDDs) and reader options (for DataFrames).
- Shuffle partitions: default 200 often under- or over-provisions. Tune:
```
--conf spark.sql.shuffle.partitions= {total_executor_cores * 2 to 3}
```
- Target ~100–256 MiB per partition after wide transforms; too small causes scheduler overhead and too big risks executor OOM.
- Shuffle is the dominant cost for joins, groupBy, and orderBy. Ensure adequate executor memory and disk; consider local SSDs for heavy shuffle on clusters.
Skew and join strategy
- Detect skew (long-tail task runtimes, large partition sizes). Mitigations:
  - Broadcast small tables to avoid shuffles:
```
--conf spark.sql.autoBroadcastJoinThreshold=64m
```
  - Salt keys for hot partitions; apply map-side pre-aggregation; filter early.
  - Enable Adaptive Query Execution (AQE) to coalesce post-shuffle partitions and handle skewed joins:
```
--conf spark.sql.adaptive.enabled=true
```
Caching, checkpointing, and lineage
- Cache hot intermediate DataFrames sparingly when reused; prefer MEMORY_AND_DISK to avoid OOM.
- Checkpoint long lineages to GCS or HDFS to bound recomputation on failures.
Executors and dynamic allocation
- Right-size executors to balance parallelism and GC overhead:
  - Cores per executor: 2–5 for balanced I/O/CPU tasks; fewer cores reduce GC pauses.
  - Memory overhead: set spark.yarn.executor.memoryOverhead for wide shuffles.
  - Enable dynamic allocation with external shuffle service on clusters to scale executors with workload:
```
--conf spark.dynamicAllocation.enabled=true
--conf spark.shuffle.service.enabled=true
--conf spark.dynamicAllocation.minExecutors=0
--conf spark.dynamicAllocation.maxExecutors=200
```
Fault-tolerance patterns for batch ETL
- Idempotent writes: write to a temp/staging path, then atomically promote with directory-level commit; for BigQuery, write to staging table and MERGE:
```
MERGE target t USING staging s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT (...)
```
- Incremental processing: use watermark-based filtering on ingestion_date partitions; maintain a processed-manifest in GCS to avoid reprocessing.
- Dead-letter handling: on parse/validation errors, branch bad records to a quarantine path/table with diagnostics. For strict schema enforcement and built-in DLQs, consider Dataflow; with Spark, implement per-record try/catch and a separate sink.

Security, Observability, and Cost

Identity and access
- Run clusters and jobs under dedicated service accounts with least-privilege IAM. Grant only required roles, for example:
  - roles/dataproc.worker to instance service accounts
  - roles/storage.objectViewer or objectAdmin for GCS I/O paths
  - roles/bigquery.dataEditor on target datasets
- For Dataproc Serverless, use per-job service accounts to scope access.
Network isolation and encryption
- Use private IP clusters in a VPC subnet, firewall-restrict master UIs, and enable Private Google Access for GCS/BigQuery without public egress.
- Place clusters in Shared VPC projects for centralized control. Optionally enable Kerberos on Dataproc for in-cluster auth.
- Encrypt at rest with CMEK: configure CMEK on GCS buckets, Persistent Disks, Dataproc Metastore, and BigQuery; use TLS in transit by default.
Logging, history, and metrics
- Enable Spark event logs to GCS and deploy the History Server:
```
--conf spark.eventLog.enabled=true
--conf spark.eventLog.dir=gs://bucket/spark-events/
```
- Dataproc streams driver and YARN logs to Cloud Logging; export to sinks for retention/forensics.
- Monitor with Cloud Monitoring metrics: YARN pending containers, CPU, memory, HDFS health (if used), GCS throughput. Alert on prolonged stage retries, executor loss, and speculative execution spikes.
- Failure analysis: common causes include skew-induced stragglers, executor OOM during shuffle, object-store commit failures, and preemptible/spot node loss. Raise retry counts judiciously; excessive retries can amplify cost and delay.
Cost optimization
- Use ephemeral clusters or Dataproc Serverless to avoid idle cost; keep data in GCS to minimize persistent disk.
- Add preemptible/spot secondary workers to absorb peak demand; design for recomputation since tasks on lost nodes are retried. Do not place masters on preemptible nodes.
- Right-size machine types and use autoscaling to shrink capacity when queues are empty. Favor Parquet/ORC with partition pruning to cut scan cost and CPU.
- Avoid small files by compacting outputs; fewer, larger files reduce metadata overhead and job runtime.
- For short, periodic jobs (for example, weekly 30-minute Spark ETL), preemptible workers or serverless often deliver the best cost profile.

Practical Problem Scenario

Acme Retail is migrating a 30-node on-prem Hadoop cluster running nightly Spark and Hive ETL that feeds downstream analytics. They want to reuse existing jobs with minimal changes, avoid managing clusters full-time, persist data beyond cluster lifetimes, and reduce storage cost.

Approach:

Land data and metadata in managed services
- Store all raw and curated data in Cloud Storage using Parquet with partitioning (for example, dt=YYYY-MM-DD).
- Rationale: GCS is durable, low-cost, and decouples compute from storage so ephemeral clusters and serverless jobs can run without persistent disks. Partitioned Parquet enables predicate pushdown and efficient scans.
Centralize the catalog with Dataproc Metastore
- Migrate the Hive metastore to Dataproc Metastore. Create external Hive tables referencing GCS paths and retain existing schema/partition logic.
- Rationale: A managed metastore allows multiple ephemeral clusters and serverless jobs to share table definitions without running an HA MySQL/PostgreSQL instance.
Use ephemeral Dataproc clusters for batch ETL and workflow templates for orchestration
- Define a workflow template that creates a cluster with the required image (for example, 2.1-debian11), runs Spark jobs (spark-sql and pyspark), and deletes the cluster on completion. Add initialization actions to install any custom libraries.
- Rationale: Ephemeral clusters eliminate idle cost and isolate job dependencies. Workflow templates provide repeatability and parameterization (dates, input paths).
Enable autoscaling and preemptible workers
- Attach an autoscaling policy with a small core worker group and a larger pool of preemptible secondary workers; tune cooldowns to scale down promptly post-run.
- Rationale: Core workers maintain cluster stability; preemptible workers absorb shuffles and wide transformations at lower cost. Spark/YARN retries handle lost tasks on preemption.
Integrate with BigQuery via the Spark BigQuery connector
- For dimension/fact loads, write Spark results to staging BigQuery tables, then run MERGE statements to update targets atomically. Where direct overwrite is safe, write partitioned tables using partition overwrite mode.
- Rationale: BigQuery serves analytics and BI at scale; staging+MERGE yields transactional-like upserts from batch Spark, reducing downstream inconsistency.
Tune Spark for performance and reliability
- Set shuffle partitions relative to executor cores and enable AQE:
```
--conf spark.sql.shuffle.partitions=600
--conf spark.sql.adaptive.enabled=true
```
- Use broadcast joins for small dimensions and checkpoint long lineages to GCS for stability.
- Rationale: Proper partitioning reduces skew and scheduler overhead; AQE adapts to data profiles at runtime; checkpointing bounds recomputation after failures.
Harden security and networking
- Run clusters with dedicated service accounts granting only roles needed for GCS paths, the metastore, and BigQuery datasets. Create private IP clusters in a restricted subnet with Private Google Access and limit UI access via firewall rules.
- Rationale: Least privilege and network isolation reduce attack surface; private control-plane egress avoids public exposure.
Instrument logging, history, and alerts
- Enable Spark event logs to GCS and deploy the History Server; route driver/YARN logs to Cloud Logging with retention. Add Monitoring alerts for long pending containers, repeated task failures, or excessive job duration.
- Rationale: Centralized logs support root-cause analysis; proactive alerts detect skew, OOMs, or degraded I/O early.
Modernize selectively with Dataproc Serverless for ad hoc and elastic spikes
- Move sporadic or exploratory Spark SQL workloads to Dataproc Serverless; keep nightly pipelines on ephemeral clusters until fully validated on serverless.
- Rationale: Serverless removes cluster ops and scales automatically, ideal for unpredictable loads; existing workflows continue with minimal code change.
Validate object-store committers and small-file management
- Set FileOutputCommitter algorithm v2 and compact outputs to 256–512 MiB per file via repartition/coalesce before writes.
- Rationale: Object stores lack atomic rename; optimized committers reduce copy/rename overhead. Compaction mitigates the small-files problem for performance and cost.

This design reuses existing Spark and Hive jobs with minimal refactoring, ensures data durability in GCS, centralizes schemas, contains security blast radius, provides robust observability, and optimizes cost through ephemeral clusters, autoscaling, preemptible capacity, and targeted use of serverless execution.

← Messaging · All domains · Data Ingestion →

Practice these questions → · Timed practice on ExamRoll.io →