CERTIFIED-ASSOCIATE-DEVELOPER-FOR-APACHE-SPARK-3-5-QUESTIONS Practice Questions

The most effective join optimization strategy in this scenario is the broadcast hash join. This strategy involves sending the smaller DataFrame (DataFrame B, 1 GB) to every executor node in the cluster. By doing this, each executor has a complete local copy of the user lookup table. The join can then be performed on each partition of the large DataFrame (DataFrame A, 128 GB) locally, without requiring the costly network operation of shuffling the massive 128 GB DataFrame. This significantly improves performance by minimizing data movement across the network.

A. The primary performance gain from a broadcast join comes from avoiding the shuffle of the large DataFrame, not the small one being broadcasted.

C. Attempting to broadcast the large DataFrame (128 GB) is incorrect. This would likely cause an OutOfMemoryError on the driver node as it tries to collect the entire large DataFrame.

D. This statement is factually incorrect; DataFrame A is the larger of the two DataFrames and should not be broadcasted.

---

1. Apache Spark Official Documentation

SQL Tuning Guide: In the section on Join Optimization

the documentation states: "Spark uses broadcast join instead of hash join when the size of one side of the join data is smaller than the spark.sql.autoBroadcastJoinThreshold... Spark broadcasts the smaller table to all executors... This avoids a shuffle of the data in the larger table." This directly supports broadcasting the smaller table (B) to avoid shuffling the larger one (A).

Source: Apache Spark 3.5.0 Documentation

Performance Tuning > SQL Tuning Guide > Join Optimization.

2. Databricks Official Documentation

Join optimization: The Databricks documentation on optimizations explains: "When you join a large table to a small table

a broadcast join can give you better performance than a merge join. A broadcast join broadcasts the small table to all nodes in the cluster and performs a hash join locally on each node

which reduces network I/O." This confirms that the smaller table is the candidate for broadcasting.

Source: Databricks Documentation > Workloads > Data engineering > Delta Lake > Optimizations > Join optimization.

3. Karau

H.

Konwinski

A.

Wendell

P.

& Zaharia

M. (2015). Learning Spark: Lightning-Fast Big Data Analysis. O'Reilly Media

Inc. In Chapter 5

"Working with Key/Value Data

" the authors explain the broadcast join mechanism: "If one of the RDDs is much smaller than the other... we can do a broadcast hash join... This avoids the cost of the shuffle." This principle is fundamental to Spark's architecture and directly applies to DataFrames.

Source: Chapter 5

section on "Joins".

The MEMORYANDDISK storage level instructs Spark to first attempt to store the DataFrame's partitions in memory as deserialized objects. If the DataFrame is too large to fit entirely in memory, Spark will store the partitions that fit and "spill" the remaining partitions to disk. When an action requires a partition stored on disk, Spark reads it from there. This process avoids job failure due to insufficient memory but introduces performance overhead because disk I/O is significantly slower than memory access. This behavior ensures that the computation can proceed even when the data exceeds available RAM.

A. Spark does not duplicate the entire DataFrame on disk. It only stores the partitions that do not fit in memory. The 2 replication levels would create duplicates, but that is not specified here.

B. The split is not "even." Spark fills memory to its capacity first and only then writes subsequent partitions to disk. It is a spill-over mechanism, not a balanced distribution.

D. The MEMORYANDDISK storage level operates on entire partitions, not on a row-by-row basis according to access frequency. It fills memory with partitions until it is full, then spills the rest.

1. Apache Spark 3.4.1 Documentation

RDD Programming Guide

"RDD Persistence" section: The official documentation describes the MEMORYANDDISK storage level as: "Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory

store the partitions that don't fit on disk

and read them from there when they're needed." This directly supports the "spill to disk" behavior.

(Source: https://spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-persistence)

2. Chambers

B.

& Zaharia

M. (2018). Spark: The Definitive Guide. O'Reilly Media

Inc. In Chapter 13

"Caching

" the section on "Storage Levels" explains that MEMORYANDDISK will "spill to disk if the RDD is too large to fit in memory." This confirms that disk is used as an overflow for data that cannot be held in memory.

3. Databricks Documentation

"Optimize performance with caching": This guide details caching strategies in Databricks. It reiterates the Spark documentation

stating for MEMORYANDDISK: "If the RDD does not fit in memory

store the partitions that don't fit on disk

and read them from there when they're needed." This confirms the behavior of storing as much as possible in memory and spilling the rest to disk.

(Source: https://docs.databricks.com/en/optimizations/cache.html)

The trigger(processingTime="5 seconds") configuration explicitly sets up a micro-batch processing model where the Spark engine initiates a new batch at a fixed interval. This means the query will check for new data and start processing it every 5 seconds, regardless of whether the previous batch has finished. This directly fulfills the requirement for a real-time analytics pipeline that processes data in micro-batches at a fixed 5-second interval.

B. trigger(continuous="5 seconds") enables continuous processing, a low-latency mode, not the requested fixed-interval micro-batch model. The interval here is for checkpointing, not batching.

C. trigger(once=True) processes only a single batch of all available data and then stops the query, which is not suitable for a continuous, real-time pipeline.

D. The default trigger (no trigger specified) starts a new micro-batch immediately after the previous one completes. This does not guarantee a fixed interval; the timing depends on processing duration.

1. Apache Spark 3.5.0 Documentation

Structured Streaming Programming Guide

Section: Triggers.

"If you specify a processing time trigger

Spark will check for new data when the previous processing has completed. If new data is available

Spark will start a new processing. If no new data is available

Spark will sleep until the processing time is reached

and then check for new data again." This describes the behavior of Trigger.ProcessingTime("5 seconds")

confirming option A. The documentation also explicitly contrasts this with the default behavior (option D)

Trigger.Once() (option C)

and Trigger.Continuous() (option B).

2. Databricks Documentation

Structured Streaming

"Configure Structured Streaming trigger intervals".

Under the heading "Fixed interval micro-batches

" the documentation states: "You can configure structured streaming queries to run with a fixed batch interval using processingTime. If you do not specify a processingTime trigger

queries run using the default micro-batch setting...". This directly validates that processingTime is the correct parameter for achieving a fixed interval.

The Iterator[Series] -> Iterator[Series] Pandas UDF is the ideal pattern for this scenario. This type of UDF is invoked once for each data partition. The function receives an iterator of pandas Series, allowing the expensive model (gettranslationmodel) to be loaded just once at the beginning of the function. The code can then iterate through the batches of data within the partition, applying the already-loaded model to each batch. This avoids the significant performance overhead of re-initializing the model for every single batch, which is what happens with a standard Series -> Series UDF.

A. Converting to a standard PySpark UDF would be less performant, as it operates row-by-row, potentially loading the model for every single row instead of for each batch.

B. A Series -> Scalar UDF still processes data in batches (pd.Series), so it would not solve the underlying problem of reloading the model for each batch.

C. While mapInPandas also uses an iterator pattern, the most direct and idiomatic way to modify an existing Pandas UDF for this optimization is to change its type to Iterator[Series] -> Iterator[Series].

1. Apache Spark Documentation

Pandas UDFs (a.k.a. Vectorized UDFs): The section on "Iterator of Series to Iterator of Series UDF" explicitly states: "It is useful when the UDF execution requires an expensive initialization... The example below shows how to use an iterator of series UDF to load a large model once and then apply it to each series in the partition." This directly addresses the question's scenario.

Source: Apache Spark 3.5.0 Documentation

pyspark.sql.functions.pandasudf

Section: "Iterator of Series to Iterator of Series UDF".

2. Databricks Documentation

User-defined functions - Python: The documentation describes the Iterator[pd.Series] -> Iterator[pd.Series] UDF and highlights its primary use case: "This is useful when the UDF execution requires an expensive state to be initialized... The state can be created once and used for the entire partition."

Source: Databricks Documentation

"User-defined functions - Python"

Section: "Pandas UDFs (vectorized UDFs)"

Subsection: "Iterator of Series to Iterator of Series".

In Apache Spark Structured Streaming, the complete output mode is specifically designed for aggregation queries. When this mode is used, the entire updated Result Table, which contains all the aggregated data from the beginning of the stream, is written to the sink during each trigger interval. This fulfills the requirement of ensuring the entire result table is outputted every time the stream processes new data.

B. append: This mode is not supported for streaming aggregations without a watermark. It only writes new rows, not the entire updated result table.

C. replace: This is a save mode for static DataFrames (df.write.mode("replace")), not a valid output mode for the writeStream API.

D. aggregate: This is a type of transformation or operation performed on a DataFrame, not a valid output mode for a streaming query.

1. Apache Spark 3.4.1 Documentation: In the Structured Streaming Programming Guide

under the "Output Modes" section

it states: "Complete Mode - The entire updated Result Table will be written to the external storage. This is supported for aggregation queries."

Source: Apache Spark Documentation

Structured Streaming Programming Guide

Output Modes

2. Databricks Documentation: The documentation on Structured Streaming output modes clarifies the behavior: "In complete mode

the entire updated results table is written to the sink on every trigger. This mode is for use cases where you need to recompute and overwrite the entire output for a given query for each batch."

Source: Databricks Documentation

Apache Spark

Structured Streaming

Output modes (Section: "Complete mode")

Apache Spark's Catalyst Optimizer automatically selects the most efficient physical plan for a query. For join operations, when one of the DataFrames is significantly smaller than the other, a Broadcast Hash Join is often the most performant strategy. This avoids a costly and network-intensive shuffle of the larger DataFrame. Spark makes this decision based on the configuration parameter spark.sql.autoBroadcastJoinThreshold, which has a default value of 10 MB. In this scenario, df2 has a size of 8 MB, which is below the default threshold. Therefore, Spark will automatically broadcast df2 to each executor node to be joined with the partitions of df1.

A. The decision to use a broadcast join is part of the initial static query plan if size estimates are available; it is not solely dependent on Adaptive Query Execution (AQE).

C. A large size difference between the DataFrames is the ideal condition for a broadcast join to be efficient, as it prevents shuffling the much larger table.

D. While a broadcast hint can force this strategy, Spark's optimizer will automatically choose a broadcast join if the size condition is met, making a hint unnecessary.

---

1. Apache Spark Official Documentation: The spark.sql.autoBroadcastJoinThreshold configuration is the key to this behavior. The documentation states its default value and purpose.

Source: Apache Spark 3.5.1 Documentation

SQL Configuration.

Reference: In the section for Performance Tuning

the description for spark.sql.autoBroadcastJoinThreshold specifies: "Configures the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join... The default value is 10M." This confirms that a table of 8 MB will be automatically broadcast.

2. Databricks Official Documentation: Databricks documentation on performance tuning explicitly details this optimization.

Source: Databricks Documentation

"Optimize performance - Join performance".

Reference: The section on "Broadcast hash join" explains: "Spark can use a broadcast join when one side of the join is small enough to fit in memory on each executor... When spark.sql.autoBroadcastJoinThreshold is enabled

Spark automatically checks if the table meets the threshold."

3. Learning Spark

2nd Edition (O'Reilly): This book

co-authored by Databricks employees and Spark committers

is a standard academic and professional reference.

Source: Damji

J. S.

Wenig

B.

Das

T.

& Lee

D. (2020). Learning Spark

2nd Edition. O'Reilly Media

Inc.

Reference: Chapter 9

"Optimizing and Tuning Spark Applications

" in the section "Join Strategies

" explains that if one DataFrame is small enough to fit in the memory of a single executor (controlled by spark.sql.autoBroadcastJoinThreshold)

the Catalyst optimizer will choose a broadcast hash join to avoid a shuffle.

To connect an existing Spark application to a Spark Connect server without code modification, the connection must be configured externally. The two primary methods are using a command-line flag when launching the application (like the PySpark shell) or setting an environment variable. The --remote "sc://localhost" flag instructs the PySpark shell to operate in Connect mode and connect to the specified server. Similarly, setting the SPARKREMOTE="sc://localhost" environment variable before launching any Spark application (including a shell or a submitted job) will configure the default SparkSession to connect to the specified remote server, overriding the default behavior of creating a local Spark session.

A. The https:// protocol is incorrect for Spark Connect; the correct protocol is sc://.

D. This option requires modifying the application's source code by adding .remote() to the SparkSession.builder call, which the question explicitly prohibits.

E. spark.connect.grpc.binding.port is a server-side configuration property that specifies the port the Spark Connect server listens on, not how a client connects to it.

1. Apache Spark 3.5.1 Documentation

Spark Connect Overview

"Start Spark Connect server and client": This section explicitly demonstrates the two valid methods.

For the command-line option: "To use Spark Connect with the PySpark shell

run the following command in the Spark root directory: ./bin/pyspark --remote sc://localhost". This validates option B.

For the environment variable: "Alternatively

you can set the SPARKREMOTE environment variable

for example

export SPARKREMOTE="sc://localhost"". This validates option C.

The documentation also shows the code-based method

.remote("sc://localhost")

confirming why option D is a valid method in general but incorrect for this specific question's constraint.

The symptoms described—low number of active tasks and fewer tasks than available CPUs—are classic indicators of under-partitioning. With 160 available CPU cores (10 nodes 16 CPUs), the cluster is significantly underutilized. The most effective and scalable approach to resolve this is to determine the number of partitions based on the data size. By dividing the total dataset size (1 TB) by a recommended partition size (e.g., 128 MB), you ensure that each task processes a manageable chunk of data. This creates a high number of partitions (approx. 8,192), allowing Spark to achieve maximum parallelism by keeping all CPU cores busy, thus optimizing resource allocation and job performance.

A. This method ignores the data size. For a 1 TB dataset, 160 partitions would be ~6.25 GB each, which is excessively large and likely to cause out-of-memory (OOM) errors.

B. Using a fixed, arbitrary value like 200 is not a best practice. It fails to account for the specific characteristics of the dataset size and cluster resources, leading to suboptimal performance.

C. This would result in only 10 partitions for 160 cores, leading to extreme under-utilization and unmanageably large partitions (~100 GB each), which would certainly fail.

1. Book: "Spark: The Definitive Guide" by Bill Chambers and Matei Zaharia (creator of Apache Spark).

Reference: Part V

Chapter 17: Performance Tuning. The chapter explains that the number of partitions is the key to parallelism. For input data from file-based sources

Spark by default creates one partition per block of the file in HDFS (defaulting to 128 MB). This establishes 128 MB as a sound

fundamental partition size. The strategy in option D directly applies this principle.

2. Book: "Learning Spark

2nd Edition" by Jules S. Damji

et al.

Reference: Chapter 10

"Performance Tuning

" section on "Number of Partitions." The authors state

"For a large dataset of 1 TB

you would want more partitions... A good starting point is to have partitions of size 128 MB." This directly endorses the calculation method described in option D as a best practice for large datasets.

3. Databricks Documentation

"Best practices: performance tuning".

Reference: In sections discussing optimization

Databricks frequently advises on ideal file and partition sizes. The general guidance is to aim for partition sizes between 128 MB and 1 GB. Option D's approach of calculating the partition count to achieve a size of 128 MB aligns perfectly with this official vendor recommendation for optimal performance.

The Parquet files are stored in a nested directory structure (/year/month/day/). By default, the Spark DataFrameReader (spark.read.parquet) only reads files from the top-level directory provided and does not search subdirectories. To instruct Spark to traverse the entire directory tree and discover all files within the nested subdirectories, the recursiveFileLookup option must be explicitly set to true. This ensures all Parquet files under /path/events/data/ are included in the resulting DataFrame, regardless of their depth in the directory hierarchy.

A. The inferSchema option is not relevant for file discovery in nested directories. It instructs Spark to deduce the schema, which is unnecessary for Parquet as the schema is stored within the file metadata.

C. Using a single wildcard () is insufficient. It would only match the first level of directories (e.g., 2023, 2024) and would not recursively descend into the month and day subdirectories to find the files.

D. This is the default behavior. It will only look for Parquet files directly inside /path/events/data/ and will fail to find any data because all files are located in subdirectories.

1. Apache Spark 3.5.1 Documentation

"Generic File Source Options": The official documentation lists recursiveFileLookup as an option for file-based data sources. It states

"If true

recursively scan the subdirectories for files. Note that this is only supported for file-based data sources like Parquet

ORC

JSON

CSV

etc."

Source: Apache Spark Documentation > SQL

DataFrames and Datasets Guide > Data Sources > Generic File Source Options

recursiveFileLookup section.

2. Databricks Documentation

"Generic file options": The Databricks documentation confirms this behavior

stating that the recursiveFileLookup option can be used to "Recursively load files and infer schema from subdirectories." This directly addresses the scenario of reading data from a nested directory structure.

Source: Databricks Documentation > Data management > Read and write data > Generic file options.

To ensure a Structured Streaming pipeline can recover from failures and continue from where it left off, you must enable checkpointing. This is achieved by specifying a path in a fault-tolerant file system (e.g., HDFS, S3, ADLS Gen2) using the checkpointLocation option on the DataStreamWriter (the object returned by writeStream). Spark uses this location to save all progress information for the query, including the range of offsets processed for each batch and the state of any running aggregations. When the query is restarted, it uses this checkpoint data to resume processing from the exact point it stopped, guaranteeing end-to-end, exactly-once fault tolerance.

A. By configuring the option checkpointLocation during readStream: The checkpointLocation option is a configuration for the data sink (writeStream), which manages the overall query progress, not the data source (readStream).

B. By configuring the option recoveryLocation during the SparkSession initialization: Fault tolerance is configured on a per-query basis, not globally on the SparkSession. Furthermore, recoveryLocation is not a valid configuration option for this purpose.

C. By configuring the option recoveryLocation during writeStream: The correct option name to specify the checkpoint directory for a streaming query is checkpointLocation, not recoveryLocation.

1. Apache Spark Official Documentation: The Structured Streaming Programming Guide explicitly states

"In case of a failure or intentional shutdown

you can recover the previous progress and state of a previous query and continue where it left off. This is done using checkpointing... you can set a checkpoint location for a query by setting the checkpointLocation option in the DataStreamWriter."

Source: Apache Spark 3.5.1 Documentation

Structured Streaming Programming Guide

Section: "Recovering from Failures with Checkpointing".

2. Databricks Official Documentation: The Databricks documentation on productionizing Structured Streaming jobs confirms this method. "To recover from failures

you must configure your streaming query with a checkpoint location... You set the path for the checkpoint using the checkpointLocation option."

Source: Databricks Documentation

Structured Streaming

Section: "Configure Structured Streaming checkpointing to recover from query failures".

3. Learning Spark

2nd Edition (O'Reilly): This book

co-authored by Databricks engineers

is a key resource. In Chapter 10

it details

"To enable this recovery

you have to set a checkpoint location for your query... You can set this location using the checkpointLocation option in the DataStreamWriter."

Source: Chambers

B.

& Zaharia

M. (2020). Learning Spark

2nd Edition. O'Reilly Media

Inc. Chapter 10

"Structured Streaming"

p. 261

Section: "Recovering from Failures with Checkpointing".