A data engineering team runs a PySpark batch transformation pipeline on an existing Google Cloud Dataproc cluster. The job reads several terabytes of raw logs stored in Google Cloud Storage and joins this massive dataset with a 15 MB dimension table containing account metadata. During job execution, the team observes two major performance bottlenecks: Cloud Storage read throughput is suboptimal during the initial data ingestion stage. The join stage triggers massive cross-worker shuffle traffic, causing slow task execution and cluster resource contention. How should the team optimize this pipeline for a specific job submission without altering cluster-wide settings?

Submit the job with `--properties=spark.hadoop.fs.gs.client.type=STORAGE_CLIENT` to enable gRPC for Cloud Storage reads, and apply a broadcast join (`broadcast()`) on the dimension table in the PySpark logic.: ### What the solution is This solution combines **connector-level transport optimization** with **Spark execution plan tuning**. Enabling the **gRPC storage client** configures the Cloud Storage connector to use Google's high-performance gRPC API instead of the standard JSON HTTP API for read operations. In parallel, wrapping the small lookup dataset in a `broadcast()` hint instructs Spark to perform a **Broadcast Hash Join (BHJ)** instead of a distributed **Sort-Merge Join (SMJ)**. ### How it meets the requirements - **Cloud Storage read performance:** Passing `--properties=spark.hadoop.fs.gs.client.type=STORAGE_CLIENT` at job submission enables **gRPC read requests** on the Cloud Storage connector specifically for this job without requiring cluster recreation or cluster-wide restarts. - **Eliminating shuffle overhead:** The 15 MB metadata table is sufficiently small to fit in worker memory. Broadcasting it distributes a local copy of the dimension table to every executor, enabling each mapper to perform joins locally against partition data and completely eliminating the **cross-network shuffle stage**. - **Job-level isolation:** Using `spark.hadoop.` property prefixes during `gcloud dataproc jobs submit` ensures the configuration applies strictly to this execution without modifying the underlying Dataproc cluster properties. ### Key advantages - **Higher read throughput:** gRPC utilizes HTTP/2 multiplexing and streaming, significantly reducing latency and CPU overhead on high-throughput Cloud Storage reads. - **Zero shuffle data transfer:** Broadcast joins bypass wide transformations, saving substantial network I/O and disk serialization overhead across executors. - **Targeted deployment:** Allows fine-tuned optimization on a per-job basis without impacting other workloads sharing the Dataproc cluster. ### Why this is optimal Alternative join strategies like sort-merge or shuffle-hash require partitioning both datasets by join key across all worker nodes. For asymmetric joins between large fact tables and tiny dimension tables, broadcast joins eliminate shuffle entirely. Pairing this with gRPC maximizes end-to-end data throughput from storage ingestion to final processing.

Optimize Dataproc Spark Grpc Broadcast Join | professional-cloud-data-engineer

Optimize Dataproc Spark Grpc Broadcast Join

Worked example. The correct answer is already marked and every option is explained below, so there is nothing to select here. To answer questions yourself, start the free trial.

A data engineering team runs a PySpark batch transformation pipeline on an existing Google Cloud Dataproc cluster. The job reads several terabytes of raw logs stored in Google Cloud Storage and joins this massive dataset with a 15 MB dimension table containing account metadata.

During job execution, the team observes two major performance bottlenecks:

Cloud Storage read throughput is suboptimal during the initial data ingestion stage.
The join stage triggers massive cross-worker shuffle traffic, causing slow task execution and cluster resource contention.

How should the team optimize this pipeline for a specific job submission without altering cluster-wide settings?

Submit the job with --properties=core:fs.gs.client.type=STORAGE_CLIENT to enable gRPC, and invoke repartition() on the dimension table before joining.

Submit the job with --properties=spark.hadoop.fs.gs.client.type=HTTP_CLIENT, and persist both the raw logs and dimension table in memory using cache().

Submit the job with --properties=spark.sql.sources.gcs.grpc.enabled=true, and set spark.sql.autoBroadcastJoinThreshold=-1 to force bucketed sort-merge joins.

Submit the job with --properties=spark.hadoop.fs.gs.client.type=STORAGE_CLIENT to enable gRPC for Cloud Storage reads, and apply a broadcast join (broadcast()) on the dimension table in the PySpark logic.

Professional Cloud Data Engineer

Optimize Dataproc Spark Grpc Broadcast Join

Related questions