Databricks-Spark-Assoc Hard Questions — 25 Difficult Questions

Question 1

Refer to the exhibit. What is the most likely cause of the repeated ExecutorLostFailure messages in the logs?

Accepted Answer

The executor process reached its memory limit and was killed.. ExecutorLostFailure indicates that an executor has died, often due to exceeding memory limits or heartbeats failing. When processing large data, the executor's heap memory can be exhausted, leading to a crash. This signifies that the cluster resources are insufficient for the current workload, requiring either more memory per executor or more efficient data processing strategies to prevent the executor from being terminated by the operating system or the cluster manager.

Answer

The driver node is experiencing a network latency issue.

Answer

The Spark job is attempting to use too many partitions.

Answer

The shuffle partition count is set too low for the data.

Question 2

You are building a Structured Streaming pipeline that reads from a Delta table source and applies a stateful deduplication using dropDuplicates on a composite key. After several hours, the job fails with an error indicating that the state store has grown too large. You need to bound the state size while still removing duplicate events that arrive within a reasonable window. Which approach should you take?

Accepted Answer

Add withWatermark on the event-time column and use dropDuplicatesWithinWatermark on the key columns.. Bounding state requires a rule that tells the engine when a key will no longer be needed. A watermark on event time provides that rule, and dropDuplicatesWithinWatermark uses it to evict keys older than the watermark. This keeps deduplication correct within the allowed lateness while preventing state from growing without limit. Configuration tweaks and repartitioning do not change the fundamental retention policy.

Answer

Set the checkpoint location to a new path and restart the query so the old state is discarded.

Answer

Increase the spark.sql.streaming.stateStore.maintenanceInterval to force more frequent cleanup of old keys.

Answer

Repartition the stream by the composite key before calling dropDuplicates to spread state across more partitions.

Question 3

A Spark application reads a large Parquet dataset, performs a filter, and then writes the result to Delta Lake. The job fails with an OutOfMemoryError on the driver. The driver heap size is set to 4 GB, and the application uses a broadcast join for a small lookup table. Which action is most likely to resolve the driver OOM while preserving the broadcast join?

Accepted Answer

Avoid using collect() or toPandas() on large datasets; write directly to Delta Lake.. Driver OOM in Spark is commonly caused by actions that bring large datasets to the driver, such as collect() or toPandas(). Writing directly to Delta Lake keeps processing distributed, avoiding driver memory overload while retaining the broadcast join. Increasing memory or changing join strategy does not address the underlying issue and may conflict with the requirement.

Answer

Replace the broadcast join with a sort-merge join.

Answer

Increase spark.driver.memory to 8 GB.

Answer

Set spark.sql.autoBroadcastJoinThreshold to -1 to disable broadcasting.

Question 4

Which THREE of the following are valid ways to create a DataFrame from an existing table in Spark SQL?

Accepted Answer

spark.table('table_name'). Spark provides multiple entry points to access table data. You can use the 'spark.table()' method for direct catalog access, 'spark.sql()' to execute a SELECT statement, or the 'spark.read.table()' method. All three provide the same underlying Dataset/DataFrame API, allowing for flexible programmatic interaction with metadata managed by the Hive Metastore or Unity Catalog, ensuring consistency across different development styles and API usage patterns in Databricks.

Answer

spark.open('table_name')

Answer

spark.from_table('table_name')

Question 5

A developer is troubleshooting a Spark job that fails with an OutOfMemoryError on the driver. The job collects a large DataFrame to the driver using .collect() and then processes it locally. The developer wants to avoid the driver OOM while still obtaining the results. Which approach is most appropriate?

Accepted Answer

Write the DataFrame to a distributed storage system and then read it back in smaller chunks for local processing.. Collecting a large DataFrame to the driver is an anti-pattern because it moves all data to a single node, risking OOM. The scalable solution is to persist the DataFrame to distributed storage and then read it in smaller partitions or batches for local processing. This maintains distribution and avoids overwhelming the driver.

Answer

Replace .collect() with .take(1000) to limit the number of rows returned.

Answer

Increase spark.driver.memory to a very large value to accommodate the collected data.

Answer

Use .foreach() to process each row on the driver instead of collecting the entire DataFrame.

Question 6

A Spark application is running on a Databricks cluster with adaptive query execution (AQE) enabled. The job reads a partitioned Parquet dataset, performs a join between two large tables, and then writes the result. The Spark UI shows that some tasks are taking significantly longer than others, and there is a high degree of skew in the join keys. Which TWO actions can help mitigate the skew and improve performance? (Choose two.)

Accepted Answer

Enable skew join optimization by setting spark.sql.adaptive.skewJoin.enabled to true.. Enabling AQE's skew join optimization and leveraging AQE's dynamic coalescing and skew handling directly address the skew issue by splitting large partitions and balancing tasks. Other options either do not target skew or are inappropriate for large tables. These two actions are the most effective for mitigating skew in this scenario.

Answer

Increase the number of shuffle partitions by setting spark.sql.shuffle.partitions to a higher value.

Answer

Repartition the larger table by the join key before the join to evenly distribute the data.

Answer

Use broadcast join by setting spark.sql.autoBroadcastJoinThreshold to a large value to broadcast the smaller table.

Question 7

A developer is using Structured Streaming with a Kafka source and wants to ensure that each message is processed exactly once, even in the event of failures. The developer has set a checkpoint location and is using `foreachBatch` to write to an external database. Which additional step is necessary to achieve exactly-once semantics?

Accepted Answer

Use the batch ID from `foreachBatch` to make writes to the external database idempotent.. Exactly-once semantics in Structured Streaming with an external sink like a database require both checkpointing and idempotent writes. The batch ID from `foreachBatch` enables idempotency by allowing you to deduplicate writes. Checkpointing ensures the stream can recover from failures without reprocessing completed batches. The other options do not address the need for idempotent writes to the external system.

Answer

Set the Kafka source option `failOnDataLoss` to true.

Answer

Configure the external database to use a transaction isolation level of SERIALIZABLE.

Answer

Enable Kafka's idempotent producer by setting `enable.idempotence=true`.

Question 8

A data engineer is using Spark SQL to join two large tables, 'orders' and 'customers', on customer_id. The 'orders' table has a column 'order_date' and the 'customers' table has a column 'signup_date'. The engineer wants to include only orders placed after the customer's signup date. Which join condition correctly implements this requirement?

Accepted Answer

orders.join(customers, orders.customer_id == customers.customer_id).filter(orders.order_date > customers.signup_date). The correct approach is to perform an equi-join on customer_id and then apply a filter for the date comparison. This works because Spark SQL supports non-equi conditions in filters after the join. Attempting to include the inequality in the join condition is not supported and will cause an error. Other options either select a boolean instead of filtering or use the wrong comparison operator.

Answer

orders.join(customers, orders.customer_id == customers.customer_id, 'inner').filter(orders.order_date > customers.signup_date)

Answer

orders.join(customers, (orders.customer_id == customers.customer_id) & (orders.order_date > customers.signup_date))

Answer

orders.join(customers, orders.customer_id == customers.customer_id).where(orders.order_date > customers.signup_date)

Question 9

A Spark Structured Streaming job on Databricks reads from a Delta table and writes micro-batches to another Delta table with a 30-second trigger. After several hours, the batch duration grows from 4 seconds to over 60 seconds and the job falls behind. The source table is compacted regularly, and the cluster has enough CPU. Which tuning action is most likely to restore the original batch duration?

Accepted Answer

Check for stateful aggregation or deduplication without a watermark and add a watermark with a bounded state cleanup so state does not grow unbounded.. A steady increase in batch duration over hours with a compacted source and adequate CPU strongly indicates unbounded state growth in a stateful streaming operation. Without a watermark, Spark retains all keys indefinitely, so each micro-batch reads and writes a larger state store. Adding a watermark with a bounded delay lets Spark evict old state, keeping per-batch work roughly constant and restoring stable batch times.

Answer

Increase spark.sql.shuffle.partitions to a very high value so each micro-batch uses more tasks and finishes faster.

Answer

Reduce the number of shuffle partitions and disable adaptive query execution so the streaming plan stays deterministic across micro-batches.

Answer

Enable Delta Lake optimized writes and tune the trigger interval so the job processes larger, less frequent micro-batches.

Question 10

A developer is building a Spark Connect client application that will run against a remote Databricks cluster. The application must handle data locally for small results and must avoid unsupported client APIs. Which two statements correctly describe behavior or limitations the developer should account for? (Choose two.)

Accepted Answer

The client can register a Python UDF with `spark.udf.register`, and the function definition is serialized to the server for execution on the cluster.. Spark Connect retains DataFrame and SQL operations and supports UDFs by shipping their definitions to the server, but it removes direct JVM and RDD access and makes result collection a client-memory concern. Designing around server-side aggregation and avoiding unsupported low-level APIs keeps the application compatible and efficient.

Answer

The client can directly manipulate the JVM through `spark._jvm` to call Java/Scala methods on the server.

Answer

`SparkContext` and RDD APIs are fully available in Spark Connect clients and can be used interchangeably with DataFrame operations.

Answer

`spark.conf.set` can be used to change arbitrary Spark runtime configurations on the remote cluster, including executor memory, without restarting the session.

Question 11

A data engineer is using Spark Connect to interact with a remote Databricks cluster. The engineer needs to understand the limitations of Spark Connect compared to traditional Spark. Which TWO of the following statements accurately describe limitations of Spark Connect? (Choose two.)

Accepted Answer

Spark Connect does not support the use of the spark-submit command to launch applications; it requires a different deployment model.. Spark Connect introduces a client-server architecture that changes how applications are developed and deployed. Two key limitations are the absence of the RDD API, which means developers must use DataFrames or Datasets, and the fact that spark-submit is not used to launch Spark Connect clients. Instead, clients are standalone applications that connect to a remote Spark server. These limitations impact migration strategies and require developers to adapt their code and deployment processes.

Answer

Spark Connect does not support the use of broadcast variables, because they cannot be serialized over the gRPC protocol.

Answer

Spark Connect does not support the use of custom JARs or libraries that are not included in the server environment.

Answer

Spark Connect requires the client to have the same version of Python as the remote cluster to avoid serialization errors.

Question 12

A developer is using Spark Connect to run a complex job that includes a Python UDF. The job fails with a serialization error. The developer suspects that the UDF is trying to use a client-side library that is not available on the cluster. Which action should the developer take to resolve this?

Accepted Answer

Install the client-side library on the Databricks cluster as a cluster library, then restart the cluster and rerun the job.. When a Python UDF in Spark Connect references a library that is not installed on the cluster, it fails with a serialization or import error. The correct resolution is to install the required library on the Databricks cluster as a cluster library. After installation, the cluster must be restarted so the library is available to the executors. This ensures the UDF can import and use the library during execution.

Answer

Use spark.addArtifact() to upload the library to the cluster, and then call spark.conf.set("spark.pyspark.driver.python", "path/to/library") to make it available.

Answer

Rewrite the UDF to avoid using the client-side library by implementing the functionality in pure PySpark built-in functions.

Answer

Add the library to the client's PYTHONPATH and set the spark.pyspark.client.pythonPath configuration to point to the client's Python environment.

Question 13

A data scientist is working with a Pandas API on Spark DataFrame `psdf` that contains a column `timestamp` of type `TimestampType`. They need to extract the day of the week and the hour of the day as new integer columns. Which two methods correctly achieve this while preserving distributed execution? (Choose two.)

Accepted Answer

`psdf['day_of_week'] = psdf['timestamp'].dt.dayofweek`. The `.dt` accessor in Pandas API on Spark provides vectorized datetime properties such as `dayofweek` and `hour` that are executed distributedly using Spark's built-in functions. These methods are efficient and preserve the lazy evaluation of the DataFrame. In contrast, using `.apply` with a Python lambda forces row-wise processing and can be prohibitively slow, while other options contain syntax errors or unnecessary transformations.

Answer

`psdf['day_of_week'] = psdf['timestamp'].transform(lambda x: x.dt.dayofweek)`

Answer

`psdf['day_of_week'] = psdf['timestamp'].apply(lambda x: x.dayofweek)`

Answer

`psdf['hour'] = psdf['timestamp'].astype('timestamp').dt.hour()`

Question 14

A data scientist is using Pandas API on Spark to analyze a large dataset. They perform a groupby operation followed by an apply of a custom function that returns a pandas Series. The operation is running very slowly and sometimes fails with out-of-memory errors on the executors. What is the most likely cause and the recommended solution?

Accepted Answer

The groupby operation is causing a shuffle, and the apply function is materializing each group in memory; they should use applyInPandas with a specified schema instead.. The groupby.apply operation in Pandas API on Spark can be inefficient because it may collect each group to a single executor and apply the function in a non-distributed manner, leading to out-of-memory errors when groups are large. Using applyInPandas with a specified schema allows for distributed execution and better memory management, as it operates on pandas DataFrames per group and returns a DataFrame, avoiding the collection of all groups to the driver. This is the recommended approach for custom group-wise operations on large datasets.

Answer

The dataset is too large for Pandas API on Spark; they should switch to using PySpark DataFrame operations exclusively.

Answer

The custom function is not vectorized; they should use built-in Pandas API on Spark functions instead of apply.

Answer

The custom function is returning a pandas Series, which is not supported; they should return a scalar instead.

Question 15

A data engineer has a DataFrame `events` with columns `user_id` and `ts`. They need to add a column `prev_ts` that holds the previous event timestamp for each user, ordered by `ts` ascending, without collapsing rows. Which operation accomplishes this?

Accepted Answer

Window.partitionBy('user_id').orderBy('ts') with lag('ts', 1) via withColumn. Using a window partitioned by `user_id` and ordered by `ts` with the `lag` function returns the previous event's timestamp for each row while retaining all rows. The other choices either aggregate away detail, rely on partition-local sorting that cannot guarantee correct ordering, or drop rows entirely. Only the windowed `lag` approach yields a per-row previous timestamp at the original granularity.

Answer

events.groupBy('user_id').agg(max('ts').alias('prev_ts'))

Answer

events.sortWithinPartitions('user_id', 'ts').withColumn('prev_ts', lead('ts'))

Answer

events.dropDuplicates(['user_id']).withColumnRenamed('ts', 'prev_ts')

Question 16

A Databricks job processing a Delta table with 50,000 small files is slow because each task opens many files. The developer wants to reduce the number of files read per task without rewriting the table. Which two actions should be taken? (Choose two.)

Accepted Answer

Enable spark.sql.adaptive.enabled and let Adaptive Query Execution coalesce post-shuffle partitions.. Adaptive Query Execution coalesces small post-shuffle partitions at runtime, cutting the number of tasks that process fragmented data, while Delta optimized writes reduce the number of files produced by shuffle writes. Together they lower file-open overhead without a manual rewrite. Settings that increase partition counts or disable broadcast joins do not consolidate small files and can make the problem worse.

Answer

Set spark.sql.files.maxPartitionBytes to a very small value so each task reads fewer bytes.

Answer

Increase spark.sql.shuffle.partitions to 4000 so more tasks can read the small files in parallel.

Answer

Set spark.sql.autoBroadcastJoinThreshold to 0 to force a broadcast join and avoid reading small files.

Question 17

Which TWO of the following statements accurately describe the role of the Spark Executor in a cluster deployment?

Accepted Answer

Executors execute the tasks dispatched by the Driver.. Executors are the workhorses of the Spark architecture, responsible for executing tasks and storing data. Recognizing their duality—compute and storage—is critical for tuning Spark applications. If executors are improperly sized, they can lead to OOM errors or underutilization of cluster resources. Understanding how executors manage task parallelism and block storage allows developers to optimize memory settings and partition counts effectively for high-performance data processing pipelines.

Answer

Executors are responsible for scheduling tasks across the worker nodes.

Answer

Executors create the physical execution plan from user code.

Answer

Executors coordinate the cluster-wide resource allocation for the job.

Question 18

A developer needs to optimize a Spark application that performs repetitive filtering and grouping on the same large DataFrame. Which feature should they implement to improve performance?

Accepted Answer

Use the cache() or persist() method on the DataFrame.. Caching (or persisting) the DataFrame keeps it in memory or on disk for subsequent actions. This is essential for iterative algorithms or workloads where the same data is reused, as it avoids recomputing the lineage from scratch. Understanding the trade-offs between memory and disk persistence is fundamental for building performant Databricks pipelines that maximize resource reuse and minimize redundant computation time.

Answer

Increase the number of cores per executor.

Answer

Switch the file format from Parquet to CSV.

Answer

Enable speculative execution.

Question 19

A streaming job reads clickstream events and performs a stateful `groupBy("sessionId").count()` with a 15-minute watermark on `eventTime`. The developer notices that state store size keeps growing without bound even though watermarks are configured. Which change most directly limits the state growth?

Accepted Answer

Ensure the watermark is defined on the event-time column and that the query is not using a non-time-based aggregation key that prevents eviction.. Stateful aggregation state is only evicted when the watermark advances beyond the event time of a key and the aggregation is keyed on that event-time. If the watermark is defined on a different column, or if the aggregation does not include the event-time window, the engine has no signal to drop old keys, so the state store grows without limit. Correcting the watermark column and key alignment is the direct remedy.

Answer

Reduce the watermark to a shorter duration, such as 1 minute, so state entries expire sooner.

Answer

Enable `spark.sql.streaming.stateStore.providerClass` to the RocksDB state store so state is written to disk instead of memory.

Answer

Switch the output mode from "update" to "append" so only finalized rows are emitted and old state is discarded.

Question 20

A developer must remove duplicate rows from a DataFrame `orders` so that only the most recent order per `customer_id` is kept, based on `order_ts`. Which TWO approaches achieve this correctly? (Choose two.)

Accepted Answer

Sort descending by `order_ts`, then use `dropDuplicates(["customer_id"])`.. A windowed `row_number` filtered to 1 gives a deterministic single latest row per customer with all columns intact, and sorting descending before `dropDuplicates` on the key captures the newest row as the survivor. Both produce one complete order per customer, whereas the tuple-based and aggregate approaches either keep stale rows or discard needed columns.

Answer

Use `dropDuplicates(["customer_id", "order_ts"])` on the DataFrame.

Answer

Use `groupBy("customer_id").agg(max("order_ts").alias("order_ts"))`.

Answer

Use `distinct()` on the DataFrame after selecting `customer_id` and `order_ts`.