Databricks Databricks-Certified-Associate-Developer-for-Apache-Spark-3.5 Dumps

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.

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.

To remove specific columns from a PySpark DataFrame, the drop() method is used. This method returns a new DataFrame without the specified columns. The correct syntax for dropping multiple columns is to pass each column name as a separate argument to the drop() method.

Correct Usage:

df_user_non_pii = df_user.drop("first_name", "last_name", "email", "birthdate")

This line of code will return a new DataFrame df_user_non_pii that excludes the specified PII columns.

Explanation of Options:

A. Correct. Uses the drop() method with multiple column names passed as separate arguments, which is the standard and correct usage in PySpark.

B. Although it appears similar to Option A, if the column names are not enclosed in quotes or if there's a syntax error (e.g., missing quotes or incorrect variable names), it would result in an error. However, as written, it's identical to Option A and thus also correct.

C. Incorrect. The dropfields() method is not a method of the DataFrame class in PySpark. It's used with StructType columns to drop fields from nested structures, not top-level DataFrame columns.

D. Incorrect. Passing a single string with comma-separated column names to dropfields() is not valid syntax in PySpark.

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.

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.

Apache Spark employs a lazy evaluation model for transformations. This means that when transformations (e.g., map(), filter()) are applied to a DataFrame, Spark does not execute them immediately. Instead, it builds a logical plan (lineage) of transformations to be applied.

Execution is deferred until an action (e.g., collect(), count(), save()) is called. At that point, Spark's Catalyst optimizer analyzes the logical plan, optimizes it, and then executes the physical plan to produce the result.

This lazy evaluation strategy allows Spark to optimize the execution plan, minimize data shuffling, and improve overall performance by reducing unnecessary computations.

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.

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.

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.

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.

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 to getOrCreate() return the same session, unless explicitly stopped.

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

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

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

(Source: PySpark SparkSession API Docs)

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.

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.

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.

Spark Connect is a client-server architecture introduced in Apache Spark 3.4, designed to decouple the client from the Spark driver, enabling remote connectivity to Spark clusters.

According to the Spark 3.5.5 documentation:

"Majority of the Streaming API is supported, including DataStreamReader, DataStreamWriter, StreamingQuery and StreamingQueryListener."

This indicates that Spark Connect supports key components of Structured Streaming, allowing for robust streaming data processing capabilities.

Regarding other options:

B. While Spark Connect supports DataFrame, Functions, and Column APIs, it does not support SparkContext and RDD APIs.

C. Spark Connect supports multiple languages, including PySpark and Scala, not just PySpark.

D. Spark Connect does not have built-in authentication but is designed to work seamlessly with existing authentication infrastructures.

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.

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

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.

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.

Watermarking in Structured Streaming defines how late a record can arrive based on event time before Spark discards it.

Behavior:

withWatermark("event_time", "10 minutes")

This means Spark will keep state for 10 minutes beyond the maximum event time seen so far.

Any data arriving later than 10 minutes after the current watermark is ignored — it will not be included in the aggregation or output.

Why the other options are incorrect:

B: Late data beyond the watermark threshold is not included.

C: Late data is not moved to a new window; it’s simply dropped.

D: True for late data within the watermark threshold, not after it.

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.

Apache Spark follows a lazy evaluation model, meaning transformations (like filter(), select(), map()) are not executed immediately. Instead, they build a logical plan (lineage graph) that represents the sequence of operations to be applied.

Execution only begins when an action (e.g., count(), collect(), save(), show()) is called. At that point, Spark’s engine:

Optimizes the logical plan into a physical plan.

Divides it into stages and tasks.

Executes them across the cluster.

This design helps Spark optimize execution paths and avoid unnecessary computations.

Why the other options are incorrect:

A: Transformations do not execute immediately; they are deferred.

B: Optimization happens during job execution (after an action), not during transformations.

D: Execution starts automatically once an action is triggered, no manual intervention needed.

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

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.

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.

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

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.

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

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.

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.

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

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.

The function date_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.

Pandas UDFs (vectorized UDFs) process entire Pandas Series objects, not scalar values. Each invocation operates on a column (Series) rather than a single value.

Correct syntax:

@pandas_udf(StringType())

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

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

This allows Spark to apply the function in a vectorized way, improving performance significantly over traditional Python UDFs.

Why the other options are incorrect:

A/D: These define scalar functions — not compatible with Pandas UDFs.

B: Uses an invalid type hint [pd.Series] (not a valid Python type annotation).

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.

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.

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.

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.

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.