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

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.

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.

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

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.

When reading Parquet files, Spark infers a unified schema automatically only if all files share identical structures.

If files have different but compatible schemas, you must enable schema merging by setting the option mergeSchema=True.

Correct syntax:

df = spark.read.option("mergeSchema", True).parquet("/data/parquet/")

This option ensures Spark merges all discovered fields across Parquet files into one unified DataFrame schema.

Why the other options are incorrect:

A: Loads files but ignores extra columns — uses only the first file’s schema.

C: inferSchema applies to CSV/JSON, not Parquet.

D: mergeAllCols is not a valid Spark option.

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

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.

dedup_df = iot_bronze_df.dropDuplicates(["event_ts", "sensor_id", "metric_value"])

dropDuplicates accepts a list of columns to use for deduplication.

This ensures only unique records based on the specified keys are retained.

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.

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.

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.

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.

Pandas API on Spark provides a distributed implementation of the Pandas DataFrame API on top of Apache Spark.

Advantages:

Executes transformations in parallel across all nodes and cores in the cluster.

Maintains Pandas-like syntax, making it easy for Python users to transition.

Enables scaling of existing Pandas code to handle large datasets without memory limits.

Therefore, it combines Pandas usability with Spark’s distributed power, offering both speed and scalability.

Why the other options are incorrect:

B: While it uses Python, that’s not its main advantage.

C: It runs distributed across the cluster, not on a single node.

D: Pandas API on Spark uses lazy evaluation, not eager computation.

In Spark’s distributed architecture:

The Driver Node coordinates the execution of a Spark application. It converts the logical plan into a physical plan of stages and tasks.

The Executors, running on Worker Nodes, are responsible for executing tasks assigned by the driver and storing data (in memory or disk) during execution.

Key point:

Executors are the active agents that perform the actual computations on data partitions. Each executor runs multiple tasks in parallel using available CPU cores.

Why the other options are incorrect:

A (Driver Nodes): The driver schedules tasks; it doesn’t execute them.

C (CPU Cores): CPU cores execute within executors, but they are hardware, not Spark architectural components.

D (Worker Nodes): Worker nodes host executors but do not directly execute tasks; executors do.

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

Spark Architecture Components — Driver, Executors, Cluster Manager, Worker Nodes.

Databricks Exam Guide (June 2025): Section “Apache Spark Architecture and Components” — describes the roles of driver and executor nodes in distributed processing.

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.

To ensure that data written out to disk is sorted, it is important to consider how Spark writes data when saving to Parquet tables. The methods .sort() or .orderBy() apply a global sort but do not guarantee that the sorting will persist in the final output files unless certain conditions are met (e.g. a single partition via .coalesce(1) — which is not scalable).

Instead, the proper method in distributed Spark processing to ensure rows are sorted within their respective partitions when written out is:

sortWithinPartitions("column_name")

According to Apache Spark documentation:

"sortWithinPartitions() ensures each partition is sorted by the specified columns. This is useful for downstream systems that require sorted files."

This method works efficiently in distributed settings, avoids the performance bottleneck of global sorting (as in .orderBy() or .sort()), and guarantees each output partition has sorted records — which meets the requirement of consistently sorted data.

Thus:

Option A and B do not guarantee the persisted file contents are sorted.

Option C introduces a bottleneck via .coalesce(1) (single partition).

Option D correctly applies sorting within partitions and is scalable.

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.

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.

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

When writing a DataFrame using .partitionBy() in Spark, the data is physically organized into directory structures corresponding to unique combinations of the partition columns.

Example:

/path/to/output/color=Red/taste=Sweet/part-0001.parquet

/path/to/output/color=Green/taste=Sour/part-0002.parquet

This structure improves query performance by pruning partitions when filtering on these columns.

Why the other options are incorrect:

A: Appending requires .mode("append"), which isn’t used here.

B: Null values in partition columns are handled; they don’t raise errors.

D: Partitioning prevents storing all data in a single file.

RDD actions like reduce() aggregate values across all partitions and return the result to the driver.

To compute the maximum processing time, reduce() is ideal because it combines results from all tasks efficiently.

Example:

max_time = rdd_times.reduce(lambda x, y: max(x, y))

This aggregates maximum values from all executors into a single result on the driver.

Why the other options are incorrect:

A: Broadcast variables distribute read-only data; they cannot aggregate results.

B: Spark UI provides visualization, not programmatic collection.

D: Accumulators support additive operations only (e.g., counters, sums), not non-associative ones like max.

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

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.

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.

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

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.

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.

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

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

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.

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.

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

Apache Spark is a distributed data processing engine designed for large-scale, cluster-based computation.

Advantages:

Horizontal Scalability: Spark can distribute tasks across many machines, handling datasets larger than the memory of a single node.

Fault Tolerance: Spark automatically recovers from node or task failures using the lineage graph (RDD recovery mechanism) and retry logic.

These two features allow Spark to process huge datasets efficiently and reliably, unlike standard Python scripts that are limited to one machine and fail on single-node errors.

Why the other options are incorrect:

B: Spark runs on commodity hardware; no specialized machines required.

C: Spark emphasizes in-memory processing, not disk-only operations.

D: Spark still requires user code in Python, Scala, SQL, or Java.

The setLogLevel() method of SparkContext sets the logging level on the driver, which controls the verbosity of logs emitted during job execution. Supported levels are inherited from log4j and include the following:

ALL

DEBUG

ERROR

FATAL

INFO

OFF

TRACE

WARN

According to official Spark and Databricks documentation:

"Valid log levels include: ALL, DEBUG, ERROR, FATAL, INFO, OFF, TRACE, and WARN."

Among the choices provided, only option B (ERROR, WARN, TRACE, OFF) includes four valid log levels and excludes invalid ones like "FAIL" or "NONE".