Databricks-Certified-Associate-Developer-for-Apache-Spark-3.5 Databricks Certified Associate Developer for Apache Spark 3.5 – Python Questions and Answers

In Spark SQL and DataFrame operations, the configuration parameter spark.sql.shuffle.partitions defines the number of partitions created during shuffle operations such as join, groupBy, and distinct.

The default value (in Spark 3.5) is 200.

If this number is too low, Spark creates fewer tasks, leading to idle executors and poor cluster utilization.

Increasing this value allows Spark to create more tasks that can run in parallel across executors, effectively using more cluster resources.

Correct approach:

spark.conf.set("spark.sql.shuffle.partitions", 400)

This increases the parallelism level of shuffle stages and improves overall resource utilization.

Why the other options are incorrect:

B: Reducing partitions further would decrease parallelism and worsen the underutilization issue.

C: Dynamic resource allocation scales executors up or down based on workload, but it doesn’t fix low task parallelism caused by insufficient shuffle partitions.

D: Increasing dataset size is not a tuning solution and doesn’t address task-level under-parallelization.

References (Databricks Apache Spark 3.5 – Python / Study Guide):

Spark SQL Configuration: spark.sql.shuffle.partitions — controls the number of shuffle partitions.

Databricks Exam Guide (June 2025): Section “Troubleshooting and Tuning Apache Spark DataFrame API Applications” — tuning strategies, partitioning, and optimizing cluster utilization.

===========

In Structured Streaming, checkpoints store state information (offsets, progress, and metadata) needed to resume a stream after a failure or restart.

Correct usage:

Set the checkpointLocation option when writing the streaming output:

streaming_df.writeStream \

format("delta") \

option("checkpointLocation", "/path/to/checkpoint/dir") \

start("/path/to/output")

Spark uses this checkpoint directory to recover progress automatically and maintain exactly-once semantics.

Why the other options are incorrect:

A/D: recoveryLocation is not a valid Spark configuration option.

B: Checkpointing must be configured in writeStream, not during readStream.

The correct answer is D because it uses proper time-based window aggregation along with watermarking, which is the required pattern in Spark Structured Streaming for time-based aggregations over event-time data.

From the Spark 3.5 documentation on structured streaming:

"You can define sliding windows on event-time columns, and use groupBy along with window() to compute aggregates over those windows. To deal with late data, you use withWatermark() to specify how late data is allowed to arrive."

(Source: Structured Streaming Programming Guide)

In option D, the use of:

python

CopyEdit

groupBy("sensor_id", window("timestamp", "5 minutes"))

agg(avg("temperature").alias("avg_temp"))

ensures that for each sensor_id, the average temperature is calculated over 5-minute event-time windows. To complete the logic, it is assumed that withWatermark("timestamp", "5 minutes") is used earlier in the pipeline to handle late events.

Explanation of why other options are incorrect:

Option A uses Window.partitionBy which applies to static DataFrames or batch queries and is not suitable for streaming aggregations.

Option B does not apply a time window, thus does not compute the rolling average over 5 minutes.

Option C incorrectly applies withWatermark() after an aggregation and does not include any time window, thus missing the time-based grouping required.

Therefore, Option D is the only one that meets all requirements for computing a time-windowed streaming aggregation.

In client mode, the driver runs on the same machine that submitted the job (often a developer’s workstation). If the driver has insufficient memory or CPU, it becomes a bottleneck.

Solution: Run the job in cluster mode.

In cluster mode, the driver runs inside the cluster on a worker node, benefiting from distributed cluster resources and improved performance for large workloads.

Why the other options are incorrect:

B: Executors handle tasks, not driver overhead.

C: May help temporarily but doesn’t scale; cluster mode is best practice.

D: Local mode runs everything on one JVM — worse for large workloads.

The method dropDuplicatesWithinWatermark() in Structured Streaming drops duplicate records based on a specified column and watermark window. The watermark defines the threshold for how late data is considered valid.

From the Spark documentation:

"dropDuplicatesWithinWatermark removes duplicates that occur within the event-time watermark window."

In this case, Spark will retain the first occurrence and drop subsequent records within the 30-minute watermark window.

Final Answer: B

Adaptive Query Execution (AQE) in Spark 3.x automatically optimizes query plans at runtime based on the actual data characteristics observed during job execution.

Key features of AQE include:

Dynamic switching of join strategies: Changes between sort-merge join and broadcast join based on actual shuffle sizes.

Coalescing shuffle partitions: Reduces small tasks and improves parallelism efficiency.

Handling skew joins: Dynamically splits large partitions to avoid data skew.

Thus, the most accurate answer describing AQE’s function is “dynamically switching join strategies.”

Why the other options are incorrect:

B: Table statistics are collected manually or by the metastore, not by AQE.

C: AQE benefits multi-stage jobs involving shuffles, not single-stage jobs.

D: Delta file optimization is handled by Databricks utilities, not AQE.

Adaptive Query Execution (AQE) is a Spark 3.x feature that dynamically optimizes query plans at runtime. One of its core features is:

Dynamically switching join strategies (e.g., from sort-merge to broadcast) based on runtime statistics.

Other AQE capabilities include:

Coalescing shuffle partitions

Skew join handling

Option A is correct.

Option B refers to statistics collection, which is not AQE's primary function.

Option C is too broad and not AQE-specific.

Option D refers to Delta Lake optimizations, unrelated to AQE.

Final Answer: A

In Apache Spark's cluster mode:

"The driver program runs on the cluster’s worker node instead of the client’s local machine. This allows the driver to be close to the data and other executors, reducing network overhead and improving fault tolerance for production jobs."

(Source: Apache Spark documentation – Cluster Mode Overview)

This deployment is ideal for production environments where the job is submitted from a gateway node, and Spark manages the driver lifecycle on the cluster itself.

Option A is partially true but less specific than D.

Option B is incorrect: the driver never executes all tasks; executors handle distributed tasks.

Option C describes client mode, not cluster mode.

In Structured Streaming, only transformation operations are allowed on streaming DataFrames. These include select(), filter(), where(), groupBy(), withColumn(), etc.

Example of supported transformation:

filtered_df = streaming_df.filter("count < 30")

However, actions such as count(), show(), and collect() are not supported directly on streaming DataFrames because streaming queries are unbounded and never finish until stopped.

To perform aggregations, the query must be executed through writeStream and an output sink.

Why the other options are incorrect:

A: count() is an action, not allowed directly on streaming DataFrames.

C: countDistinct() is a stateful aggregation, not supported outside of a proper streaming query.

D: show() is also an action, unsupported on streaming queries.

To compare a column value with a Python literal constant in a DataFrame expression, use F.lit() to convert it into a Spark literal.

Correct refactor:

from pyspark.sql import functions as F

min_price = 110.50

result_df = prices_df.filter(F.col("price") > F.lit(min_price)).agg(F.count("*"))

This avoids type mismatches and ensures Spark executes the filter expression on the cluster.

Why the other options are incorrect:

B: where() syntax is valid, but F.lit("price") is incorrect — wraps string literal, not a column.

C: withColumn adds a column, not needed for this aggregation.

D: Comparison logic reversed.

To sort a Spark DataFrame by multiple columns, use .orderBy() (or .sort()) with column expressions.

Correct syntax for descending and ascending mix:

from pyspark.sql.functions import col

df1.orderBy(col("count").desc(), col("Name").asc())

This sorts primarily by count in descending order and secondarily by Name in ascending order (alphabetically).

Why the other options are incorrect:

B/C: Default sort order is ascending; won’t place highest counts first.

D: Reverses sorting logic — sorts Name descending, not required.

The method saveAsTable() creates a new table and optionally fails if the table exists.

From Spark documentation:

"The mode 'ErrorIfExists' (default) will throw an error if the table already exists."

Thus:

Option A is correct.

Option B (Overwrite) would overwrite existing data — not acceptable here.

Option C and D use save(), which doesn't create a managed table with metadata in the metastore.

Final Answer: A

To process data in fixed micro-batch intervals, use the .trigger(processingTime="interval") option in Structured Streaming.

Correct usage:

query = df.writeStream \

outputMode("append") \

trigger(processingTime="5 seconds") \

start()

This instructs Spark to process available data every 5 seconds.

Why the other options are incorrect:

B: continuous triggers are for continuous processing mode (different execution model).

C: once=True runs the stream a single time (batch mode).

D: Default trigger runs as fast as possible, not fixed intervals.

If each of the 10 nodes runs 4 executors, and each executor is assigned 4 CPU cores:

Executors per node = 4

Cores per executor = 4

Total executors = 4 * 10 = 40

Total cores = 40 executors * 4 cores = 160 cores

However, Spark uses 1 core for overhead on each node when managing multiple executors. Thus, the practical allocation is:

Total usable executors = 4 executors/node × 10 nodes = 40

Total cores = 4 cores × 40 executors = 160

Answer: A — but the question asks specifically about “CPU cores used by the application,” assuming all allocated cores are usable (as Spark typically runs executors without internal core reservation unless explicitly configured).

However, if you are considering 4 executors/node × 4 cores = 16 cores per node, across 10 nodes, that’s 160.

Final Answer: A

Shuffling occurs in operations like groupBy, reduceByKey, or join—which cause data to be moved across partitions. The repartition() operation can also cause a shuffle, but in this context, it follows an aggregation.

In Option D, the groupBy followed by agg results in a shuffle due to grouping across nodes.

The repartition(10) is a partitioning transformation but does not involve a new shuffle since the data is already grouped.

This sequence — shuffle (groupBy) followed by non-shuffling (repartition) — is correct.

Option A does the opposite: the filter does not cause a shuffle, but groupBy does — this makes it the wrong order.

Spark Connect enables decoupling of the client and Spark driver processes, allowing remote access. Spark supports configuring the remote Spark Connect server in multiple ways:

From Databricks and Spark documentation:

Option B (--remote "sc://localhost") is a valid command-line argument for the pyspark shell to connect using Spark Connect.

Option C (setting SPARK_REMOTE environment variable) is also a supported method to configure the remote endpoint.

Option A is incorrect because Spark Connect uses the sc:// protocol, not https:// .

Option D requires modifying the code, which the question explicitly avoids.

Option E configures the port on the server side but doesn’t start a client connection.

Final Answers: B and C

Spark Connect introduces a decoupled client-server architecture. Its key feature is enabling Spark job submission and execution from remote clients — in Python, Java, etc.

From Databricks documentation:

“Spark Connect allows remote clients to connect to a Spark cluster and execute Spark jobs without being co-located with the Spark driver.”

A is close, but "any language" is overstated (currently supports Python, Java, etc., not literally all).

B refers to REST, which is not Spark Connect's mechanism.

D is incorrect; Spark Connect isn’t focused on ingestion.

Final Answer: C

In the Spark architecture, the driver node is responsible for orchestrating the execution of a Spark application. It converts user-defined transformations and actions into a logical plan, optimizes it into a physical plan, and then splits the plan into tasks that are distributed to the executor nodes.

As per Databricks and Spark documentation:

“The driver node is responsible for maintaining information about the Spark application, responding to a user's program or input, and analyzing, distributing, and scheduling work across the executors.”

This means:

Option A is correct because the driver schedules and coordinates the job execution.

Option B is incorrect because the driver does more than just UI monitoring.

Option C is incorrect since data and computations are distributed across executor nodes.

Option D is incorrect; results are returned to the driver but not stored long-term by it.

To read from a specific Kafka topic using Structured Streaming, the correct syntax is:

python

CopyEdit

option("subscribe", "feed")

This is explicitly defined in the Spark documentation:

"subscribe – The Kafka topic to subscribe to. Only one topic can be specified for this option."

(Source: Apache Spark Structured Streaming + Kafka Integration Guide)

B. "subscribe.topic" is invalid.

C. "kafka.topic" is not a recognized option.

D. "topic" is not valid for Kafka source in Spark.

The correct way to aggregate information (e.g., max value) from distributed workers back to the driver is using RDD actions such as reduce() or aggregate().

From the documentation:

“To perform global aggregations on distributed data, actions like reduce() are commonly used to collect summaries such as min/max/avg.”

Accumulators (Option B) do not support max operations directly and are not intended for such analytics.

Broadcast (Option C) is used to send data to workers, not collect from them.

Spark UI (Option D) is a monitoring tool — not an analytics collection interface.

Final Answer: A

Operations that trigger data movement across partitions (like groupBy, join, repartition) result in a shuffle and a new stage.

From Spark documentation:

“groupBy and aggregation cause data to be shuffled across partitions to combine rows with the same key.”

Option A (groupBy + agg) → causes shuffle.

Options B, C, and D (filter, withColumn, select) → transformations that do not require shuffling; they are narrow dependencies.

Final Answer: A

The correct output mode for streaming aggregations that need to output the full updated results at each trigger is "complete".

From the official documentation:

“complete: The entire updated result table will be output to the sink every time there is a trigger.”

This is ideal for aggregations, such as counts or averages grouped by a key, where the result table changes incrementally over time.

append: only outputs newly added rows

replace and aggregate: invalid values for output mode

The groupBy() operation causes a shuffle because it requires all values for a specific key to be brought together, which may involve moving data across partitions.

In contrast:

filter() and select() are narrow transformations and do not cause shuffles.

coalesce() tries to reduce the number of partitions and avoids shuffling by moving data to fewer partitions without a full shuffle (unlike repartition()).

dropDuplicates() with no column list removes duplicates based on all columns.

It's the most efficient and semantically correct way to deduplicate records that are completely identical across all fields.

From the PySpark documentation:

dropDuplicates(): Return a new DataFrame with duplicate rows removed, considering all columns if none are specified.

— Source: PySpark DataFrame.dropDuplicates() API

To create a Python dictionary from a Spark DataFrame, you can first collect the data to the driver node and then convert it into a Python dictionary using dict().

Steps:

Select only relevant columns.

Order by region_id to get the smallest ones.

Limit to 3 rows.

Map each row into key–value pairs.

Collect results to the driver and convert to a dictionary.

Correct code:

regions_dict = dict(

regions.orderBy("region_id")

limit(3)

rdd.map(lambda x: (x.region_id, x.region_name))

collect()

)

This produces a dictionary like:

{10: 'North', 12: 'East', 14: 'West'}

Why the other options are incorrect:

A/B: take(3) returns a list of Row objects, not key–value pairs.

C: Doesn’t order or limit by smallest IDs, so the result may not be correct.

Underutilization with timeout warnings often indicates insufficient parallelism — meaning there aren’t enough executors to process all tasks concurrently.

Solution:

Increase the number of executors to allow more parallel task execution and better resource utilization.

Example configuration:

--conf spark.executor.instances=8

This distributes the workload more effectively across cluster nodes and reduces idle time for pending tasks.

Why the other options are incorrect:

A: Extending timeouts hides the symptom, not the root cause (lack of executors).

B: More memory per executor won’t fix scheduling bottlenecks.

C: Reducing partition size may increase overhead and does not fix resource imbalance.

f.to_date() converts a timestamp or string to a DateType.

Ideal for extracting the date component (year-month-day) from a full timestamp.

Example:

from pyspark.sql.functions import to_date

dates_df.withColumn("date", to_date("timestamp"))

When you convert a large pyspark.pandas (aka Pandas API on Spark) DataFrame to a local Pandas DataFrame using .toPandas(), Spark collects all partitions to the driver.

From the Spark documentation:

“Be careful when converting large datasets to Pandas. The entire dataset will be pulled into the driver’s memory.”

Thus, for large datasets, this can cause memory overflow or out-of-memory errors on the driver.

Final Answer: D

Comprehensive Explanation:

To cover structured data processing, SQL querying, and machine learning in Apache Spark, the correct combination of components is:

Spark DataFrames: for structured data processing

Spark SQL: to execute SQL queries over structured data

MLlib: Spark's scalable machine learning library

This trio is designed for exactly this type of use case.

Why other options are incorrect:

A: GraphX is for graph processing — not needed here.

B: Pandas API on Spark is useful, but MLlib is essential for ML, which this option omits.

C: Spark Streaming is legacy; GraphX is irrelevant here.

The number of tasks is controlled by the number of partitions. By default, spark.sql.shuffle.partitions is 200. If stages are showing very few tasks (less than total cores), you may not be leveraging full parallelism.

From the Spark tuning guide:

"To improve performance, especially for large clusters, increase spark.sql.shuffle.partitions to create more tasks and parallelism."

Thus:

A is correct: increasing shuffle partitions increases parallelism

B is wrong: it further reduces parallelism

C is invalid: increasing dataset size doesn’t guarantee more partitions

D is irrelevant to task count per stage

Final Answer: A

The SparkSession is the entry point to Spark functionality in modern versions (2.x and later). It unifies the SparkContext, SQLContext, and HiveContext into a single object.

Best Practice:

Use one SparkSession for the entire application.

Create it once at the start using SparkSession.builder.getOrCreate().

Reuse it across all transformations and actions.

Stop it only after all operations are completed.

Example:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("MyApp").getOrCreate()

# Perform transformations and actions

spark.stop()

Why the other options are incorrect:

B: SparkSession is the recommended interface; SparkContext alone is deprecated for SQL/DataFrame APIs.

C: Creating multiple sessions increases overhead and wastes resources.

D: Restarting SparkSession breaks lineage and adds unnecessary startup costs.

broadcast(df1) tells Spark to send the small DataFrame (df1) to all worker nodes.

This eliminates the need for shuffling df1 during the join.

Broadcast joins are optimized for scenarios with one large and one small table.

Partitioning a dataset divides data into separate directories based on partition column values. When queries filter on partitioned columns, Spark can prune irrelevant partitions — meaning it only reads files that match the filter criteria.

Advantage:

Reduces I/O and improves performance by scanning only relevant subsets of data.

Example:

/data/sales/year=2023/month=10/...

/data/sales/year=2024/month=01/...

A query filtering WHERE year = 2024 reads only the relevant partition.

Why the other options are incorrect:

A: Compression is independent of partitioning.

B: Spark does not automatically clean partitions unless managed manually.

C: Partitioning does not cause Spark to load entire data into memory.

To extract unique values from a column and sort them alphabetically:

distinct() is required to remove duplicate values.

orderBy() is needed to sort the results alphabetically (ascending by default).

Correct code:

df.select("Name").distinct().orderBy(df["Name"])

This is directly aligned with standard DataFrame API usage in PySpark, as documented in the official Databricks Spark APIs. Option A is incorrect because it may not remove duplicates. Option C omits sorting. Option D sorts in descending order, which doesn't meet the requirement for alphabetical (ascending) order.

In Spark, when a CSV row does not match the provided schema, Spark does not raise an error by default. Instead, it returns null for fields that cannot be parsed correctly.

In the first row, "hello" cannot be cast to Integer for the age field → Spark sets age=None

In the second row, "20" is a valid integer → age=20

So the output will be:

[Row(name='bambi', age=None), Row(name='alladin', age=20)]

Final Answer: C

The message "GC overhead limit exceeded" typically indicates that the JVM is spending too much time in garbage collection with little memory recovery. This suggests that the driver or executor is under-provisioned in memory.

The most effective remedy is to increase the driver memory using:

--driver-memory 4g

This is confirmed in Spark's official troubleshooting documentation:

“If you see a lot of GC overhead limit exceeded errors in the driver logs, it’s a sign that the driver is running out of memory.”

— Spark Tuning Guide

Why others are incorrect:

A may help but does not directly address the driver memory shortage.

B is not a valid action; GC cannot be disabled.

D increases memory usage, worsening the problem.

To create a new calculated column in Spark, use the .withColumn() method.

To remove an unwanted column, use the .drop() method.

Correct syntax:

from pyspark.sql.functions import col

employees_df = employees_df.withColumn("tax", col("salary") * 0.1).drop("age")

.withColumn("tax", col("salary") * 0.1) → adds a new column where tax = 10% of salary.

.drop("age") → removes the age column from the DataFrame.

Why the other options are incorrect:

B: lit(0.1) creates a constant value, not a calculated tax.

C: .dropField() is not a DataFrame API method (used only in struct field manipulations).

D: Adds 0.1 to salary instead of calculating 10%.