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

The Spark configuration spark.executor.cores defines how many concurrent tasks can be executed within a single executor process.

Each executor is assigned a number of CPU cores.

Each core executes one task at a time.

Therefore, increasing spark.executor.cores allows an executor to run more tasks concurrently.

Example:

--conf spark.executor.cores=4

→ Each executor can run 4 parallel tasks.

Why the other options are incorrect:

B (spark.task.maxFailures): Sets retry attempts for failed tasks.

C (spark.executor.memory): Sets executor memory, not concurrency.

D (spark.sql.shuffle.partitions): Defines number of shuffle partitions, not executor concurrency.

Exactly-once guarantees in Spark Structured Streaming require micro-batch mode (default), not continuous mode.

Continuous mode (.trigger(continuous=...)) only supports at-least-once semantics and lacks full fault-tolerance.

trigger(availableNow=True) is a batch-style trigger, not suited for low-latency streaming.

So:

Option A uses micro-batching with a tight trigger interval → minimal latency + exactly-once guarantee.

Final Answer: A

To read all files recursively within a nested directory structure, Spark requires the recursiveFileLookup option to be explicitly enabled. According to Databricks official documentation, when dealing with deeply nested Parquet files in a directory tree (as shown in this example), you should set:

df = spark.read.option("recursiveFileLookup", "true").parquet("/path/events/data/")

This ensures that Spark searches through all subdirectories under /path/events/data/ and reads any Parquet files it finds, regardless of the folder depth.

Option A is incorrect because while it includes an option, inferSchema is irrelevant here and does not enable recursive file reading.

Option C is incorrect because wildcards may not reliably match deep nested structures beyond one directory level.

Option D is incorrect because it will only read files directly within /path/events/data/ and not subdirectories like /2023/01/01.

Databricks documentation reference:

"To read files recursively from nested folders, set the recursiveFileLookup option to true. This is useful when data is organized in hierarchical folder structures" — Databricks documentation on Parquet files ingestion and options.

In PySpark, when performing a join between two DataFrames, the columns do not have to share the same name. You can explicitly provide a join condition by comparing specific columns from each DataFrame.

This syntax is correct and fully supported:

df1.join(df2, df1.employee_id == df2.emp_id, how='inner')

This will perform an inner join between df1 and df2 using the employee_id from df1 and emp_id from df2.

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

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"))

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.

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.

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.

Adaptive Query Execution (AQE) dynamically selects join strategies based on actual data sizes at runtime. If the size of post-shuffle partitions is below the threshold set by:

spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold

then Spark prefers to use a shuffled hash join.

From the Spark documentation:

“AQE selects a shuffled hash join when the size of post-shuffle data is small enough to fit within the configured threshold, avoiding more expensive sort-merge joins.”

Therefore:

A is wrong — Cartesian joins are only used with no join condition.

B is correct — this is the optimized join for small partitioned shuffle data under AQE.

C and D are used under other scenarios but not for this case.

Final Answer: B

According to Spark’s watermarking rules:

“Records that are older than the watermark (event time < current watermark) are considered too late and are dropped.”

So, if a record’s event_time is earlier than (max event_time seen so far - 10 minutes), it is discarded.

spark.executor.cores determines how many concurrent tasks an executor can run.

For example, if set to 4, each executor can run up to 4 tasks in parallel.

Other settings:

spark.task.maxFailures controls task retry logic.

spark.driver.cores is for the driver, not executors.

spark.executor.memory sets memory limits, not task concurrency.

Adaptive Query Execution (AQE) is a Spark SQL feature introduced to dynamically optimize queries at runtime based on actual data statistics collected during execution.

Key benefits include:

Runtime plan adaptation: Spark adjusts the physical plan after some stages complete.

Skew handling: Automatically splits skewed partitions to balance work distribution.

Join strategy optimization: Dynamically switches between shuffle join and broadcast join depending on partition sizes.

Coalescing shuffle partitions: Reduces the number of small tasks for better performance.

Example configuration:

spark.conf.set("spark.sql.adaptive.enabled", True)

This enables AQE globally in Spark 3.5.

Why the other options are incorrect:

A: AQE adapts during runtime, not only before execution.

B: Task distribution is a base Spark feature, not specific to AQE.

C: AQE specifically addresses runtime skew and join adjustments.

The question requires creating a dictionary where keys are region values and values are the corresponding region_id integers. Furthermore, it asks to retrieve only the smallest 3 region_id values.

Key observations:

select('region', 'region_id') puts the column order as expected by dict() — where the first column becomes the key and the second the value.

sort('region_id') ensures sorting in ascending order so the smallest IDs are first.

take(3) retrieves exactly 3 rows.

Wrapping the result in dict(...) correctly builds the required Python dictionary: { 'AFRICA': 0, 'AMERICA': 1, 'ASIA': 2 }.

Incorrect options:

Option B flips the order to region_id first, resulting in a dictionary with integer keys — not what's asked.

Option C uses .limit(3) without sorting, which leads to non-deterministic rows based on partition layout.

Option D sorts in descending order, giving the largest rather than smallest region_ids.

Hence, Option A meets all the requirements precisely.

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.

A broadcast join is a type of join where the smaller DataFrame is replicated (broadcast) to all worker nodes in the cluster. This avoids shuffling the large DataFrame across the network.

Benefits:

Eliminates shuffle for the smaller dataset.

Greatly improves performance when one side of the join is small enough to fit in memory.

Correct usage example:

df_result = df2.join(broadcast(df1), "id")

This is a map-side join, where each executor joins its local partition of the large dataset with the broadcasted copy of the small one.

Why the other options are incorrect:

A: Broadcasting does not change partition sizes.

B: Joins always match on key equality; this is not specific to broadcast joins.

D: Broadcasting does not filter; it distributes data for faster joins.

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

Adaptive Query Execution (AQE) is a powerful optimization framework introduced in Apache Spark 3.0 and enabled by default since Spark 3.2. It dynamically adjusts query execution plans based on runtime statistics, leading to significant performance improvements. The key benefits of AQE include:

Dynamic Join Strategy Selection: AQE can switch join strategies at runtime. For instance, it can convert a sort-merge join to a broadcast hash join if it detects that one side of the join is small enough to be broadcasted, thus optimizing the join operation .

Handling Skewed Data: AQE detects skewed partitions during join operations and splits them into smaller partitions. This approach balances the workload across tasks, preventing scenarios where certain tasks take significantly longer due to data skew .

Coalescing Post-Shuffle Partitions: AQE dynamically coalesces small shuffle partitions into larger ones based on the actual data size, reducing the overhead of managing numerous small tasks and improving overall query performance .

These runtime optimizations allow Spark to adapt to the actual data characteristics during query execution, leading to more efficient resource utilization and faster query processing times.

Option B uses Spark’s built-in SQL function length(), which is efficient and avoids the overhead of a Python UDF:

from pyspark.sql.functions import length, col

df.select(length(col("stringColumn")).alias("length"))

Explanation of other options:

Option A is incorrect syntax; spark.udf is not called this way.

Option C registers a UDF but doesn’t apply it in the DataFrame transformation.

Option D is syntactically valid but uses a Python UDF which is less efficient than built-in functions.

Final Answer: B

The correct way to load specific columns from an ORC file is to first load the file using .load() and then apply .select() on the resulting DataFrame. This is valid with .read.format("orc") or the shortcut .read.orc().

df = spark.read.format("orc").load("/file/test_data.orc").select("col1", "col2")

Why others are incorrect:

A performs selection after filtering, but doesn’t match the intention to minimize memory at load.

B incorrectly tries to use .select() before .load(), which is invalid.

C uses a non-existent .selected() method.

D correctly loads and then selects.

Option B is the correct and idiomatic approach in PySpark to split a string column (like email) based on a delimiter such as "@".

The split(col("email"), "@") function returns an array with two elements: username and domain.

getItem(0) retrieves the first part (username).

getItem(1) retrieves the second part (domain).

withColumn() is used to create new columns from the extracted values.

Example from official Databricks Spark documentation on splitting columns:

from pyspark.sql.functions import split, col

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

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

⚠️ Why other options are incorrect:

A uses fixed substring indices (substr(0, 5)), which won’t correctly extract usernames and domains of varying lengths.

C uses substring_index, which is available but less idiomatic for splitting emails and is slightly less readable.

D removes "@" from the email entirely, losing the separation between username and domain, and ends up duplicating values in both fields.

Therefore, Option B is the most accurate and reliable solution according to Apache Spark 3.5 best practices.

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.

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

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

In Apache Spark’s execution hierarchy, the relationships are structured as follows:

Job: Created when an action (e.g., count(), collect(), save()) is triggered on an RDD or DataFrame.

Stage: Each job is divided into one or more stages, separated by shuffle boundaries (e.g., after a reduceByKey or join).

Task: Each stage consists of multiple tasks, one per partition, executed in parallel on executors.

Execution Hierarchy:

Job → Stage(s) → Task(s)

So, a job contains multiple stages, and each stage contains multiple tasks.

Why the other options are incorrect:

A: A job does not directly contain tasks without stages.

B: A stage cannot contain multiple jobs; it belongs to a single job.

C: Tasks do not contain jobs.

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

Spark Architecture Overview — Execution Hierarchy: Jobs, Stages, and Tasks.

Databricks Exam Guide (June 2025): Section “Apache Spark Architecture and Components” — describes execution hierarchy and lazy evaluation.

===========

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

To flatten an array of structs into individual rows and access fields within each struct, you must:

Use explode() to expand the array so each struct becomes its own row.

Access the struct fields via dot notation (e.g., record_exploded.sensor_id).

Option C does exactly that:

First, explode the record array column into a new column record_exploded.

Then, access fields of the struct using the dot syntax in select.

This is standard practice in PySpark for nested data transformation.

Final Answer: C

The correct answer is B because it uses the new function count_if, introduced in Spark 3.5.0, which simplifies conditional counting within aggregations.

F.count_if(condition) counts the number of rows that meet the specified boolean condition.

In this example, it directly counts how many times spot_price >= min_price evaluates to true, replacing the older verbose combination of when/otherwise and filtering or summing.

Official Spark 3.5.0 documentation notes the addition of count_if to simplify this kind of logic:

“Added count_if aggregate function to count only the rows where a boolean condition holds (SPARK-43773).”

Why other options are incorrect or outdated:

A uses a legacy-style method of adding a flag column (when().otherwise()), which is verbose compared to count_if.

C performs a simple min/max aggregation—useful but unrelated to conditional array operations or the updated functionality.

D incorrectly applies .filter() after .agg() which will cause an error, and misuses string "min_price" rather than the variable.

Therefore, B is the only option leveraging new functionality from Spark 3.5.0 correctly and efficiently.

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.

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

Both cache() and persist() are Spark DataFrame storage operations that store computed results in memory (and optionally on disk) to speed up subsequent actions on the same DataFrame.

Key difference:

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

persist() allows specifying different storage levels, such as MEMORY_ONLY, DISK_ONLY, or MEMORY_AND_DISK_SER.

Example:

df.cache() # Uses MEMORY_AND_DISK by default

df.persist(StorageLevel.MEMORY_ONLY) # Custom storage level

Both trigger caching upon an action (e.g., count(), collect()).

Why the other options are incorrect:

A: persist() default is not DISK_ONLY; default storage level is MEMORY_AND_DISK.

B/C: cache() cannot set arbitrary levels; only persist() can.

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.

When two DataFrames have different column sets, the normal union() or unionAll() functions fail unless both have exactly the same columns in the same order.

Solution: Use unionByName() with allowMissingColumns=True.

This aligns columns by name and automatically adds missing columns with null values.

Correct syntax:

combined_df = au_df.unionByName(nz_df, allowMissingColumns=True)

This ensures the union works even if one DataFrame has extra or missing columns.

Why the other options are incorrect:

B: unionAll() is deprecated; also requires identical schemas.

C: With allowMissingColumns=False, Spark still throws a mismatch error.

D: union() doesn’t accept the allowMissingColumns argument.

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

In Structured Streaming, a watermark defines the maximum delay for event-time data to be considered in stateful operations like deduplication or window aggregations.

Behavior:

df = df.withWatermark("event_timestamp", "30 minutes")

This sets a 30-minute watermark, meaning Spark will only keep track of events that arrive within 30 minutes of the latest event time seen so far. When used with:

df.dropDuplicates(["event_id", "event_timestamp"])

Spark removes duplicates that arrive within the watermark threshold (in this case, within 30 minutes).

Why other options are incorrect:

A: Watermarks do not remove all duplicates; they only manage those within the defined event-time window.

B: Watermark durations can be expressed as strings like "30 minutes", "10 seconds", etc., not only seconds.

D: Structured Streaming supports deduplication using withWatermark() and dropDuplicates().

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

PySpark Structured Streaming Guide — withWatermark() and dropDuplicates() methods for event-time deduplication.

Databricks Certified Associate Developer for Apache Spark Exam Guide (June 2025): Section “Structured Streaming” — Topic: Streaming Deduplication with and without watermark usage.