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

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.

Pandas API on Spark (pyspark.pandas) allows interoperability with PySpark DataFrames. To convert a pyspark.pandas.DataFrame to a standard PySpark DataFrame, you use .to_spark().

Example:

df = psdf.to_spark()

This is the officially supported method as per Databricks Documentation.

Incorrect options:

B, D: Invalid or nonexistent methods.

C: Converts to a local pandas DataFrame, not a PySpark DataFrame.

The provided code defines a Pandas UDF of type Series-to-Series, where a new instance of the language model is created on each call, which happens per batch. This is inefficient and results in significant overhead due to repeated model initialization.

To reduce the frequency of model loading, the engineer should convert the UDF to an iterator-based Pandas UDF (Iterator[pd.Series] -> Iterator[pd.Series]). This allows the model to be loaded once per executor and reused across multiple batches, rather than once per call.

From the official Databricks documentation:

“Iterator of Series to Iterator of Series UDFs are useful when the UDF initialization is expensive… For example, loading a ML model once per executor rather than once per row/batch.”

— Databricks Official Docs: Pandas UDFs

Correct implementation looks like:

python

CopyEdit

@pandas_udf("string")

def translate_udf(batch_iter: Iterator[pd.Series]) -> Iterator[pd.Series]:

model = get_translation_model(target_lang='es')

for batch in batch_iter:

yield batch.apply(model)

This refactor ensures the get_translation_model() is invoked once per executor process, not per batch, significantly improving pipeline performance.

df.cache() is shorthand for df.persist(StorageLevel.MEMORY_AND_DISK)

df.persist() allows specifying any storage level such as MEMORY_ONLY, DISK_ONLY, MEMORY_AND_DISK_SER, etc.

By default, persist() uses MEMORY_AND_DISK, unless specified otherwise.

The split() function in PySpark splits strings into an array based on a given delimiter.

Then, .getItem(index) extracts a specific element from the array.

Correct usage:

from pyspark.sql.functions import split, col

customerDF = customerDF \

withColumn("username", split(col("email"), "@").getItem(0)) \

withColumn("domain", split(col("email"), "@").getItem(1))

This creates two new columns derived from the email field:

"username" → text before @

"domain" → text after @

Why the other options are incorrect:

B: regexp_replace only replaces text; does not split into multiple columns.

C: .select() cannot alias multiple derived columns like this.

D: Column objects are not native Python strings; cannot use standard .split().

When running in local mode (e.g., local[4]), the number inside the brackets defines how many threads Spark will use.

Using local[*] ensures Spark uses all available CPU cores for parallelism.

Example:

spark-submit --master local[*]

Dynamic allocation and executor memory apply to cluster-based deployments, not local mode.

Temp views like createOrReplaceTempView are session-scoped.

They disappear once the Spark session ends.

To retain data across sessions, it must be persisted:

df.write.saveAsTable("users_vw")

Thus, the view needs to be persisted as a table to survive session termination.

To persist a table and specify the save path, use:

users.write.option("path", "/some/path").saveAsTable("default_table")

The .option("path", ...) must be applied before calling saveAsTable.

Option A uses invalid syntax (write(path=...)).

Option B applies .option() after .saveAsTable()—which is too late.

Option D uses incorrect syntax (no path parameter in saveAsTable).

In Spark SQL, to convert a UTC timestamp to another timezone, you use the function from_utc_timestamp().

Correct syntax:

from pyspark.sql.functions import from_utc_timestamp, col

df_converted = raw.withColumn(

"transaction_datetime",

from_utc_timestamp(col("transaction_datetime"), "America/New_York")

)

This adjusts the UTC time into the specified timezone using Spark’s timezone database.

Why the other options are incorrect:

B: to_utc_timestamp() converts local time to UTC, not the other way around.

C: date_format() formats timestamps as strings but doesn’t adjust timezones.

D: convert_timezone() is not a valid Spark SQL function.

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()).

To filter rows based on a condition and display them in Spark, use filter(...).show():

employees_df.filter(employees_df.tenure >= 5).show()

Option A is correct and shows the results.

Option B filters but doesn’t display them.

Option C uses Python’s built-in filter, not Spark.

Option D collects the results to the driver, which is unnecessary if .show() is sufficient.

Final Answer: A

To handle JSON files with evolving or differing schemas, Spark SQL supports the option mergeSchema 'true', which merges all fields across files into a unified schema.

Correct syntax:

CREATE EXTERNAL TABLE users

USING json

OPTIONS (path '/data/input.json', mergeSchema 'true');

This creates an external table directly on the JSON data, inferring schema automatically and merging variations.

Why the other options are incorrect:

B: Missing schema merge configuration — fails with inconsistent files.

C: inferSchema applies to CSV/other file types, not JSON.

D: mergeAll is not a valid Spark SQL option.

Apache Spark was designed for big data and high-velocity workloads. Its core strength lies in its in-memory computation and parallel distributed processing model.

These features allow Spark to:

Process large-scale datasets quickly across many nodes.

Support real-time and near–real-time analytics for tasks like fraud detection and recommendations.

Minimize disk I/O through caching and memory persistence.

Thus, the key advantage in this use case is Spark’s ability to handle large data volumes efficiently using distributed, in-memory computation.

Why the other options are incorrect:

A: Spark is optimized for large, not small, datasets.

C: SQL support is useful but doesn’t solve the scalability issue.

D: MLlib supports machine learning but relies on Spark’s parallel computation for speed.

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.

The most efficient and correct approach is to define a schema using StructType and pass it to spark.read.schema(...).

This avoids schema inference overhead and ensures proper data types are enforced during read.

Example:

from pyspark.sql.types import StructType, StructField, StringType, DoubleType

schema = StructType([

StructField("order_id", StringType(), True),

StructField("amount", DoubleType(), True),

])

df = spark.read.schema(schema).json("path/to/json")

— Source: Databricks Guide – Read JSON with predefined schema

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.

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.

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.

When writing DataFrames to files using the Spark DataFrameWriter API, Spark by default raises an error if the target path already exists. To explicitly overwrite existing data, you must specify the write mode as "overwrite".

Correct Syntax:

df.write.mode("overwrite").json("path/to/file")

This command removes the existing file or directory at the specified path and writes the new output in JSON format.

Other supported save modes include:

"append" — Adds new data to existing files.

"ignore" — Skips writing if the path already exists.

"error" or "errorifexists" — Fails the job if the output path exists (default).

Why other options are incorrect:

A: Defaults to "error" mode, which fails if the path exists.

B: "append" only adds data; it does not overwrite existing data.

C: .option("overwrite") is invalid — mode("overwrite") must be used instead.

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

PySpark API Reference: DataFrameWriter.mode() — describes valid write modes including "overwrite".

PySpark API Reference: DataFrameWriter.json() — method to write DataFrames in JSON format.

Databricks Certified Associate Developer for Apache Spark Exam Guide (June 2025): Section “Using Spark DataFrame APIs” — Reading and writing DataFrames using save modes, schema management, and partitioning.

===========

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’s client mode, the driver runs on the local machine that submitted the job. If that machine is resource-constrained (e.g., low memory), performance degrades.

From the Spark documentation:

"In cluster mode, the driver runs inside the cluster, benefiting from cluster resources and scalability."

Option A is incorrect — executors do not help the driver directly.

Option B might help short-term but does not scale.

Option C is correct — switching to cluster mode moves the driver to the cluster.

Option D (local mode) is for development/testing, not production.

Final Answer: C

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

Spark uses lazy evaluation. Transformations like filter, select, and groupBy only define the DAG (Directed Acyclic Graph). No execution occurs until an action is triggered.

The first action in the code is:reduced_df.count()

So Spark starts processing data at this line.

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

To write a structured streaming DataFrame to Parquet files, the correct way to specify the format and output directory is:

writeStream

format("parquet")

option("path", "path/to/destination/dir")

According to Spark documentation:

“When writing to file-based sinks (like Parquet), you must specify the path using the .option("path", ...) method. Unlike batch writes, .save() is not supported.”

Option A incorrectly uses .option("location", ...) (invalid for Parquet sink).

Option B incorrectly sets the format via .option("format", ...), which is not the correct method.

Option C repeats the same issue.

Option D is correct: .format("parquet") + .option("path", ...) is the required syntax.

Final Answer: D

Pandas API on Spark (formerly Koalas) offers:

Familiar Pandas-like syntax

Distributed execution using Spark under the hood

Scalability for large datasets across the cluster

It provides the power of Spark while retaining the productivity of Pandas.

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.

Spark automatically uses a broadcast hash join when one side of the join is small enough to fit within the broadcast threshold.

Default threshold:

spark.sql.autoBroadcastJoinThreshold = 10MB (as of Spark 3.5)

Since df2 is 8 MB, Spark automatically broadcasts it to all executors. This avoids a shuffle on the large dataset (df1) and speeds up the join.

Why the other options are incorrect:

A: 8 MB < 10 MB threshold → broadcast join is efficient.

B: AQE is not required for broadcast joins; it’s a static optimization.

C: Broadcast hints are optional — Spark infers automatically.

When using the MEMORY_AND_DISK storage level, Spark attempts to cache as much of the DataFrame in memory as possible. If the DataFrame does not fit entirely in memory, Spark will store the remaining partitions on disk. This allows processing to continue, albeit with a performance overhead due to disk I/O.

As per the Spark documentation:

"MEMORY_AND_DISK: It stores partitions that do not fit in memory on disk and keeps the rest in memory. This can be useful when working with datasets that are larger than the available memory."

— Perficient Blogs: Spark - StorageLevel

This behavior ensures that Spark can handle datasets larger than the available memory by spilling excess data to disk, thus preventing job failures due to memory constraints.

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%.

The Spark UI is the standard and most effective way to inspect executor logs, task time, input size, and shuffles.

From the Databricks documentation:

“You can monitor job execution via the Spark Web UI. It includes detailed logs and metrics, including task-level execution time, shuffle reads/writes, and executor memory usage.”

(Source: Databricks Spark Monitoring Guide)

Option A is incorrect: logs are not guaranteed to be in /tmp, especially in cloud environments.

B. —verbose helps during job submission but doesn't give detailed executor logs.

D. spark-sql is a CLI tool for running queries, not for inspecting logs.

Hence, the correct method is using the Spark UI → Stages tab → Executor logs.

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.

When queries frequently filter on certain columns, partitioning by those columns ensures partition pruning, allowing Spark to scan only relevant directories instead of the entire dataset.

Correct code:

final.write.partitionBy("event_year", "event_month").parquet("events.liveLatest")

This improves read performance dramatically for filters like:

SELECT * FROM events.liveLatest WHERE event_year = 2024 AND event_month = 5;

bucketBy() helps in clustering and joins, not in partition pruning for file-based tables.

Why the other options are incorrect:

B: Bucket count changes parallelism, not query pruning.

C: sortBy organizes data within files, not across partitions.

D: Partitioning by only one column limits pruning benefits.

To define a micro-batch interval, the correct syntax is:

query = df.writeStream \

outputMode("append") \

trigger(processingTime='5 seconds') \

start()

This schedules the query to execute every 5 seconds.

Continuous mode (used in Option A) is experimental and has limited sink support.

Option D is incorrect because processingTime must be a string (not an integer).

Option B triggers as fast as possible without interval control.

na.drop(how='any') drops any row that has at least one null value.

This is exactly what's needed when the goal is to retain only fully complete records.

Usage:CopyEdit

filtered_df = users_raw_df.na.drop(how='any')

Explanation of incorrect options:

A: thresh=0 is invalid — thresh must be ≥ 1.

B: how='all' drops only rows where all columns are null (too lenient).

D: spark.na.drop doesn't support mixing how and thresh in that way; it's incorrect syntax.

In Apache Spark, broadcast variables are used to efficiently distribute large, read-only data to all worker nodes. However, broadcasting very large datasets can lead to memory issues on executors if the data does not fit into the available memory.

According to the Spark documentation:

"Broadcast variables allow the programmer to keep a read-only variable cached on each machine rather than shipping a copy of it with tasks. This can greatly reduce the amount of data sent over the network."

However, it also notes:

"Using the broadcast functionality available in SparkContext can greatly reduce the size of each serialized task, and the cost of launching a job over a cluster. If your tasks use any large object from the driver program inside of them (e.g., a static lookup table), consider turning it into a broadcast variable."

But caution is advised when broadcasting large datasets:

"Broadcasting large variables can cause out-of-memory errors if the data does not fit in the memory of each executor."

Therefore, if the broadcasted DataFrame containing millions of rows exceeds the memory capacity of the executors, the job may fail due to memory constraints.

Which operation is supported with streaming_df?

A. streaming_df.select(countDistinct("Name"))

B. streaming_df.groupby("Id").count()

C. streaming_df.orderBy("timestamp").limit(4)

D. streaming_df.filter(col("count") < 30).show()

Answer: B

In Structured Streaming, only a limited subset of operations is supported due to the nature of unbounded data. Operations like sorting (orderBy) and global aggregation (countDistinct) require a full view of the dataset, which is not possible with streaming data unless specific watermarks or windows are defined.

Review of Each Option:

A. select(countDistinct("Name"))

Not allowed — Global aggregation like countDistinct() requires the full dataset and is not supported directly in streaming without watermark and windowing logic.