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

The .coalesce(numPartitions) function is used to reduce the number of partitions in a DataFrame. It does not increase the number of partitions. If the specified number of partitions is greater than the current number, it will not have any effect.

From the official Spark documentation:

“coalesce() results in a narrow dependency, e.g. if you go from 1000 partitions to 100 partitions, there will not be a shuffle, instead each of the 100 new partitions will claim one or more of the current partitions.”

However, if you try to increase partitions using coalesce (e.g., from 10 to 20), the number of partitions remains unchanged.

Hence, df.coalesce(20) will still return a DataFrame with 10 partitions.

When converting a standard PySpark UDF to a Pandas UDF for performance optimization, the function must operate on a Pandas Series as input and return a Pandas Series as output.

In this case, the original function signature:

def shake_256(raw: str) - > str

is scalar — not compatible with Pandas UDFs.

According to the official Spark documentation:

“Pandas UDFs operate on pandas.Series and return pandas.Series . The function definition should be:

def my_udf(s: pd.Series) - > pd.Series:

and it must be registered using pandas_udf(...) .”

Therefore, to fix the error:

The function should be updated to:

def shake_256(df: pd.Series) - > pd.Series:

return df.apply(lambda x: hashlib.shake_256(x.encode()).hexdigest(20))

This will allow Spark to efficiently execute the Pandas UDF in vectorized form, improving performance compared to standard UDFs.

The Spark History Server provides a web UI for viewing past completed applications , including event logs, stages, and performance metrics.

If disabled:

Spark jobs still run normally,

But users lose the ability to review historical job metrics , DAGs, or logs after completion.

Thus, debugging, performance analysis, and audit capabilities are lost.

Why the other options are incorrect:

A: Disabling History Server doesn’t manage logs.

B/D: Minimal overhead; disabling doesn’t improve runtime speed or executor performance.

To create an external table and enable schema merging, the correct syntax is:

CREATE EXTERNAL TABLE users

USING json

OPTIONS (

path ' /data/input.json ' ,

mergeSchema ' true '

)

mergeSchema is the correct option key (not schemaMerge )

EXTERNAL allows Spark to query files without managing their lifecycle

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.

To enable fault tolerance and ensure that Spark can resume from the last committed offset after failure, you must configure a checkpoint location using the correct option key: " checkpointLocation " .

From the official Spark Structured Streaming guide:

“To make a streaming query fault-tolerant and recoverable, a checkpoint directory must be specified using .option( " checkpointLocation " , " /path/to/dir " ) .”

Explanation of options:

Option A uses an invalid option name: " checkpoint " (should be " checkpointLocation " )

Option B is correct: it sets checkpointLocation properly

Option C lacks checkpointing and won’t resume after failure

Option D also lacks checkpointing configuration

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

To use a Python function as a UDF (User Defined Function) in Spark SQL, it must first be registered using spark.udf.register().

Correct usage:

spark.udf.register( " cube_func " , cube_func)

num_df.selectExpr( " cube_func(num) " ).show()

This registers cube_func as a callable SQL function available in expressions or queries.

Why the other options are incorrect:

B: You must wrap with udf() or selectExpr; calling plain Python functions won’t work.

C: createDataFrame is for building DataFrames, not calling UDFs.

D: DataFrames cannot directly register UDFs.

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

To enable a Structured Streaming query to recover from failures or intentional shutdowns, it is essential to specify the checkpointLocation option during the writeStream operation. This checkpoint location stores the progress information of the streaming query, allowing it to resume from where it left off.

According to the Databricks documentation:

" You must specify the checkpointLocation option before you run a streaming query, as in the following example:

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

toTable( " catalog.schema.table " )

— Databricks Documentation: Structured Streaming checkpoints

By setting the checkpointLocation during writeStream , Spark can maintain state information and ensure exactly-once processing semantics, which are crucial for reliable streaming applications.

In Pandas API on Spark (previously Koalas) , the method .to_spark() converts a pyspark.pandas.DataFrame into a PySpark DataFrame .

Correct usage:

spark_df = pdf.to_spark()

This enables interoperability between the Pandas API on Spark and the PySpark SQL API , allowing developers to switch seamlessly between both for transformations or performance optimization.

Why the other options are incorrect:

A (to_pandas) : Converts to a local Pandas DataFrame , not a PySpark DataFrame.

C (to_dataframe) : Not a valid API method.

D (spark) : Not an existing DataFrame method.

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

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.

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.

To sort a PySpark DataFrame by multiple columns with mixed sort directions, the correct usage is:

python

CopyEdit

df.orderBy( " age " , " salary " , ascending=[ True , False ])

age will be sorted in ascending order

salary will be sorted in descending order

The orderBy() and sort() methods in PySpark accept a list of booleans to specify the sort direction for each column.

Documentation Reference: PySpark API - DataFrame.orderBy

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.

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.

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

The default broadcast join threshold in Spark is:

spark.sql.autoBroadcastJoinThreshold = 10MB

Since df2 is only 8 MB (less than 10 MB), Spark will automatically apply a broadcast join without requiring explicit hints.

From the Spark documentation:

“If one side of the join is smaller than the broadcast threshold, Spark will automatically broadcast it to all executors.”

A is incorrect because Spark does support auto broadcast even with static plans.

B is correct: Spark will automatically broadcast df2.

C and D are incorrect because Spark ' s default logic handles this optimization.

Final Answer: B

To view the schema of a Parquet file, you must use the DataFrameReader to load the Parquet data and call the .printSchema() method.

Correct syntax:

spark.read.parquet( " events.parquet " ).printSchema()

This command loads the file metadata (without triggering a full read) and prints the column names, data types, and nullability information in a tree format.

Why the other options are incorrect:

A/D: SQL queries can’t directly introspect file schemas.

B: .show() displays data rows, not schema.

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 approx_percentile function in Spark is a performance-optimized alternative to percentile. It takes an optional accuracy parameter:

approx_percentile(column, percentage, accuracy)

Higher accuracy values → more precise results, but increased memory/computation.

Lower values → faster but less accurate.

From the documentation:

“Increasing the accuracy improves precision but increases memory usage.”

Final Answer: D

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.

A Spark job is triggered by an action (e.g., count, show).

The job is broken into stages, typically one per shuffle boundary.

Each stage is divided into multiple tasks, which are distributed across worker nodes.

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 approach to perform a parallelized groupBy operation across Spark worker nodes using Pandas API is via applyInPandas . This function enables grouped map operations using Pandas logic in a distributed Spark environment. It applies a user-defined function to each group of data represented as a Pandas DataFrame.

As per the Databricks documentation:

" applyInPandas() allows for vectorized operations on grouped data in Spark. It applies a user-defined function to each group of a DataFrame and outputs a new DataFrame. This is the recommended approach for using Pandas logic across grouped data with parallel execution. "

Option A is correct and achieves this parallel execution.

Option B ( mapInPandas ) applies to the entire DataFrame, not grouped operations.

Option C uses built-in aggregation functions, which are efficient but not customizable with Pandas logic.

Option D creates a scalar Pandas UDF which does not perform a group-wise transformation.

Therefore, to run a groupBy with parallel Pandas logic on Spark workers, Option A using applyInPandas is the only correct answer.

To read each file as a single record, use:

spark.read.text(path, wholetext= True )

This ensures that Spark reads the entire file contents into one row.

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.

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.

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.

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.

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.

Structured Streaming supports three output modes:

Append: Writes only new rows since the last trigger.

Update: Writes only updated rows.

Complete: Writes the entire result table after every trigger execution.

For aggregations like groupBy().count(), only complete mode outputs the entire table each time.

Example:

aggDF.writeStream \

outputMode( " complete " ) \

format( " console " ) \

start()

Why the other options are incorrect:

A: “AGGREGATE” is not a valid output mode.

C: “REPLACE” does not exist.

D: “APPEND” writes only new rows, not the full table.

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.

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 .mode( " overwrite " ) ensures that existing files at the path will be replaced.

partitionBy( " country " ) optimizes queries by writing data into partitioned folders.

Correct syntax:

df.write.mode( " overwrite " ).partitionBy( " country " ).parquet( " /data/output " )

— Source: Spark SQL, DataFrames and Datasets Guide