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

Comprehensive and Detailed Explanation From Exact Extract:

The functiondate_sub(start, days)subtracts the number of days from the start date. If a negative number is passed, the behavior becomes a date addition.

Example:

SELECT date_sub('2024-05-01', -5)

-- Returns: 2024-05-06

So, a negative value effectively adds the absolute number of days to the date.

Comprehensive and Detailed Explanation From Exact Extract:

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

From the Spark 3.5 documentation on structured streaming:

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

(Source:Structured Streaming Programming Guide)

In optionD, the use of:

python

CopyEdit

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

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

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

Explanation of why other options are incorrect:

Option AusesWindow.partitionBywhich applies to static DataFrames or batch queries and is not suitable for streaming aggregations.

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

Option Cincorrectly applieswithWatermark()after an aggregation and does not include any time window, thus missing the time-based grouping required.

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

Comprehensive and Detailed Explanation From Exact Extract:

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

Useexplode()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 therecordarray column into a new columnrecord_exploded.

Then, access fields of the struct using the dot syntax inselect.

This is standard practice in PySpark for nested data transformation.

Final Answer: C

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.

Comprehensive and Detailed Explanation From Exact Extract:

Apache Arrow is used under the hood to optimize conversion between Pandas and PySpark DataFrames. The correct configuration setting is:

spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")

From the official documentation:

“This configuration must be enabled to allow for vectorized execution and efficient conversion between Pandas and PySpark using Arrow.”

Option B is correct.

Options A, C, and D are invalid config keys and not recognized by Spark.

Final Answer: B

The correct way to overwrite an existing file using the DataFrameWriter is:

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

Option D is also technically valid, but Option A is the most concise and idiomatic PySpark syntax.

Comprehensive and Detailed Explanation:

The original code has several issues:

It references a variable answer that is undefined.

The function is annotated to return a str, but the logic attempts numeric multiplication.

The UDF return type is declared as T.IntegerType() but the function performs a floating-point operation, which is incompatible.

Option B correctly:

Uses DoubleType to reflect the fact that the multiplication involves a float (3.14159).

Declares the input as float, which aligns with the multiplication.

Returns a float, which matches both the logic and the schema type annotation.

This structure aligns with how PySpark expects User Defined Functions (UDFs) to be declared:

"To define a UDF you must specify a Python function and provide the return type using the relevant Spark SQL type (e.g., DoubleType for float results)."

Example from official documentation:

from pyspark.sql.functions import udf

from pyspark.sql.types import DoubleType

@udf(returnType=DoubleType())

def multiply_by_pi(x: float) -> float:

return x * 3.14159

This makes Option B the syntactically and semantically correct choice.

Comprehensive and Detailed Explanation From Exact Extract:

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.

Comprehensive and Detailed Explanation From Exact Extract:

The provided code defines a Pandas UDF of type Series-to-Series, where a new instance of the language modelis 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 theget_translation_model()is invoked once per executor process, not per batch, significantly improving pipeline performance.

Comprehensive and Detailed Explanation From Exact Extract:

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

In the first row,"hello"cannot be cast to Integer for theagefield → Spark setsage=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

Comprehensive and Detailed Explanation From Exact Extract:

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

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

By default,persist()usesMEMORY_AND_DISK, unless specified otherwise.

Comprehensive and Detailed Explanation From Exact Extract:

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

Comprehensive and Detailed Explanation:

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

Comprehensive and Detailed Explanation From Exact Extract:

According to the PySpark API documentation:

"getOrCreate(): Gets an existing SparkSession or, if there is no existing one, creates a new one based on the options set in this builder."

This means Spark maintains a global singleton session within a JVM process. Repeated calls togetOrCreate()return the same session, unless explicitly stopped.

Option A is incorrect: the method does not destroy any session.

Option B incorrectly ties uniqueness toappName, which does not influence session reusability.

Option D is incorrect: it contradicts the fundamental behavior ofgetOrCreate().

(Source:PySpark SparkSession API Docs)

Comprehensive and Detailed Explanation From Exact Extract:

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

Temp views likecreateOrReplaceTempVieware 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.

Comprehensive and Detailed Explanation From Exact Extract:

Spark's best practice is to estimate partition count based on data volume and a reasonable partition size — typically 128 MB to 256 MB per partition.

With 1 TB of data: 1 TB / 128 MB ≈ ~8000 partitions

This ensures that tasks are distributed across available CPUs for parallelism and that each task processes an optimal volume of data.

Option A (equal to cores) may result in partitions that are too large.

Option B (fixed 200) is arbitrary and may underutilize the cluster.

Option C (nodes) gives too few partitions (10), limiting parallelism.

Comprehensive and Detailed Explanation From Exact Extract:

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

From the Spark documentation:

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

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

Final Answer: D

Comprehensive and Detailed Explanation From Exact Extract:

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.—verbosehelps during job submission but doesn't give detailed executor logs.

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

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

Comprehensive and Detailed Explanation From Exact Extract:

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.

Comprehensive and Detailed Explanation From Exact Extract:

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