Troubleshooting and Tuning DataFrame Apps practice questions

The data is not partitioned correctly for a broadcast join.

Why it fails: Broadcast joins do not require specific partitioning because the entire table is sent to every executor. The failure is not about partitioning logic, but rather the inability of the executor's memory heap to accommodate the large amount of data sent via the broadcast variable in the network.

The driver node lacks sufficient memory to store the broadcasted table.

Why it fails: While the driver coordinates the broadcast, the table is ultimately stored in the executor's memory. The failure happens on the executors when they attempt to deserialize or load the broadcasted data. Therefore, the driver's memory is not the direct cause of this specific failure during the broadcast join.

The tables are not indexed by the join key.

Why it fails: Spark does not use indices in the traditional relational sense for join optimization. Broadcast joins are chosen based on table size heuristics, not indices. Lacking an index would not cause a join failure, as Spark will perform a full scan of the data regardless of the presence of indices.

Using a higher number of shuffle partitions.

Why it fails: Increasing partitions helps parallelism, but if one key contains the majority of the data, that key will still fall into a single task. Increasing partitions does not split the data associated with a single key, so the skew remains, and the task imbalance will persist despite the higher partition count.

Increasing the executor memory limit.

Why it fails: Increasing executor memory allows the process to handle more data, but it does not fix the underlying distribution problem. The skewed partition will still be processed by a single task, meaning the bottleneck remains, even if the container itself does not fail due to memory exhaustion during the task execution.

Setting the join algorithm to merge sort.

Why it fails: Spark's shuffle join, often implemented as a sort-merge join, is the standard for joining large datasets. Manually selecting an algorithm does not resolve skew, as the sort process itself will be bottlenecked by the partition that contains the massive amount of data associated with the skewed join keys.

Calling the repartition() method on the DataFrame.

Why it fails: Repartitioning triggers a full data shuffle across the cluster to change the distribution of data. It is an expensive operation that has nothing to do with predicate pushdown or optimizing the I/O read process from external storage. It is used to balance data distribution, not to filter input data.

Caching the DataFrame before filtering.

Why it fails: Caching forces the data into memory. If you filter after caching, the entire original dataset must be read into memory first. This negates the benefits of predicate pushdown, as you end up reading and storing the entire dataset in memory instead of just the subset you actually need for processing.

Using the collect() action on the unfiltered DataFrame.

Why it fails: The collect action retrieves all data to the driver, forcing a full scan and transfer of the entire dataset. This is the opposite of predicate pushdown, as it ensures no filtering occurs at the source and puts maximum pressure on the network and the driver's memory resources.

spark.executor.cores

Why it fails: While increasing executor cores can improve parallelism, it also increases the number of concurrent tasks competing for the same JVM heap space. This often exacerbates OOM errors rather than solving them, as memory contention between concurrent tasks increases without an equivalent increase in total available executor memory.

spark.driver.maxResultSize

Why it fails: The maxResultSize configuration limits the total size of collected results sent back to the driver node. While it can cause OOM on the driver, it does not directly resolve shuffle-related OOM errors occurring on executor nodes during transformations. It is a driver-side constraint, not an executor memory constraint.

spark.sql.autoBroadcastJoinThreshold

Why it fails: This threshold controls whether a join should be broadcast. While setting it too high can cause OOM errors by attempting to load massive tables into memory, it is not a primary configuration to adjust when troubleshooting shuffle-based OOM issues, which usually result from standard sort-merge join operations.

Decrease the number of executor instances in the cluster.

Why it fails: Decreasing the number of executors reduces the total memory available to the Spark application. This forces more tasks onto each remaining executor, which will likely worsen the memory pressure and increase the frequency of spilling to disk, leading to even slower performance and potential task failures.

Switch from Parquet to CSV format to reduce serialization overhead.

Why it fails: CSV is a text-based format that is significantly less efficient than Parquet. Switching to CSV increases data size and I/O overhead, which would actually increase memory usage during reading and processing. This would likely cause more frequent spilling to disk, not less, as the data volume effectively grows.

Disable the Spark SQL Adaptive Query Execution feature.

Why it fails: Adaptive Query Execution helps optimize shuffle partitions dynamically based on runtime statistics. Disabling it removes a crucial optimization layer that can help mitigate memory pressure. Keeping it enabled allows Spark to coalesce partitions or adjust plans to better fit available memory, which is beneficial for avoiding spilling.

Using too many shuffle partitions in the Spark configuration.

Why it fails: While too many shuffle partitions can lead to excessive small file issues and overhead, they do not cause data skew. Skew is a property of the data distribution, not the configuration of the number of partitions. In fact, more partitions can sometimes mitigate the severity of existing skew.

The presence of a high-cardinality join key column.

Why it fails: High-cardinality columns are generally ideal for join keys. Because the values are unique or spread evenly, they naturally distribute data across partitions effectively. This avoids the hotspots associated with skew and allows Spark to maintain an even balance of work across the cluster nodes.

It displays the live system memory usage of the driver and executors.

Why it fails: The 'Executors' tab provides memory metrics for the nodes, not the 'SQL' tab. The SQL tab is dedicated to the analysis of logical and physical query plans, transformation chains, and the performance of SQL or DataFrame-based operations, rather than raw JVM resource utilization metrics.

It automatically generates a list of code refactoring suggestions for the script.

Why it fails: The Spark UI is a monitoring and diagnostic tool. It does not perform static analysis or suggest code refactoring, such as changing join types or optimizing filter predicates. These decisions must be made by the developer based on the performance metrics observed in the UI's visualizations.

It allows for real-time modification of the Spark SQL execution engine code.

Why it fails: The SQL tab is a read-only monitoring interface. You cannot modify the Spark execution engine or recompile code through the UI. Any changes to the execution logic must be performed in the application code or through Spark configuration settings before submitting the job to the cluster.

Increase spark.driver.maxResultSize.

Why it fails: This parameter limits the size of the result that can be collected to the driver. It has no impact on executor-side operations like sorting. If a sort operation fails, it is due to executor-side memory pressure, which cannot be addressed by changing the driver's result size limit.

Set spark.memory.fraction to 1.0.

Why it fails: Setting memory fraction to 1.0 would allocate all memory to execution/storage, leaving zero room for user-defined objects. This usually leads to more frequent garbage collection issues and does not effectively solve sort-related memory failures, as it ignores the necessary overhead for internal Spark operations.

Disable spark.sql.adaptive.enabled.

Why it fails: Adaptive Query Execution (AQE) is designed to optimize query plans at runtime, including handling skew and partition sizing. Disabling it would remove these optimizations, likely making performance worse rather than solving memory-related failures during sort operations that require effective data management.

Increase the number of shuffle partitions.

Why it fails: Increasing shuffle partitions increases the number of output tasks, which usually exacerbates the small files problem. If each task writes a small amount of data, you end up with even more tiny files, creating significant performance bottlenecks during subsequent reads and increasing metadata management costs.

Use repartition() to increase parallelism.

Why it fails: Repartitioning increases the number of partitions, which creates more output files. If the original issue is too many small files, using repartitioning will only make the situation worse. It is the opposite of the desired outcome for consolidating data for efficient storage.

Increase spark.sql.shuffle.partitions to 4000 so the shuffle of the 2 GB table is parallelized across more tasks.

Why it fails: Raising shuffle partitions only spreads an already-required shuffle across more tasks; it does not eliminate the network transfer. The smaller table would still be exchanged between executors for every join stage. More partitions can even increase scheduling overhead and produce many tiny output files, so this does not solve the root cause.

Convert the 2 GB Parquet table to ORC and rely on predicate pushdown to remove the shuffle during the join.

Why it fails: Predicate pushdown filters rows at read time and does not influence how Spark physically executes a join. Changing the file format to ORC may improve scan efficiency, but the join operator still requires an exchange unless a broadcast join is selected. The network shuffle of the smaller table would remain unchanged.

Set spark.sql.autoBroadcastJoinThreshold to -1 to disable broadcast joins, then repartition both tables by the join key.

Why it fails: Setting the threshold to -1 disables broadcast joins entirely, which forces a SortMergeJoin and guarantees the 2 GB table is shuffled. Repartitioning both sides by the join key adds another full shuffle before the join. This makes the scenario worse, not better, and does not address the observed network transfer of the smaller table.

Set spark.sql.adaptive.enabled to false.

Why it fails: Disabling Adaptive Query Execution (AQE) would prevent the engine from dynamically coalescing shuffle partitions or optimizing skew joins. AQE can automatically reduce partition sizes or handle skew, which helps with spilling. Turning it off would remove these automatic optimizations and likely degrade performance further in this scenario.

Decrease spark.sql.shuffle.partitions to a smaller number.

Why it fails: Reducing the number of shuffle partitions creates larger partitions, which would increase the amount of data each task must handle. This would likely worsen spilling because tasks would need even more memory to process their partitions. It is counterproductive when the goal is to reduce per-task memory usage.

Increase spark.executor.memory to a larger value.

Why it fails: While adding executor memory can help with spilling, it is not the most targeted change for a high-cardinality groupBy. Simply increasing memory may not solve the issue if the number of partitions is too low, leading to oversized partitions. Adjusting shuffle partitions directly addresses the partition size, which is the root cause of spilling in this context.

Increase spark.sql.broadcastTimeout to 1800 seconds to allow more time for broadcasting.

Why it fails: The broadcastTimeout controls how long the driver waits for the broadcast to complete on each executor; it does not reduce memory usage. Increasing it may help if the job fails with a timeout, but here the symptom is out-of-memory, not timeout. This change would not address the root cause and could even delay failure detection, making debugging harder.

Disable broadcast join by setting spark.sql.autoBroadcastJoinThreshold to -1.

Why it fails: Disabling broadcast join forces a sort-merge join, which requires a full shuffle of both tables. For a 500 GB fact table, this shuffle is extremely expensive and will not resolve the memory pressure; it will likely make the job slower and introduce new bottlenecks. The broadcast side is only 2 GB, so the memory issue probably stems from too many concurrent broadcast copies or insufficient executor memory per core.

Repartition the large table to 2000 partitions before the join.

Why it fails: Repartitioning the large table increases parallelism but does not reduce the memory required for the broadcast table on each executor. In fact, more partitions can lead to more concurrent tasks per executor, each holding a broadcast copy, potentially worsening memory pressure. The issue is the broadcast side's memory footprint, not the large table's partitioning.

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

Why it fails: Lowering maxPartitionBytes creates more, smaller input partitions, which increases the number of tasks and typically worsens the small-file problem. The goal is to reduce per-task file opens, not to split input more finely. This setting is useful for controlling partition size for large files, not for consolidating many small files.

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

Why it fails: Raising shuffle partitions increases parallelism but also multiplies the number of output partitions and files, aggravating the small-file problem. Each of the 4,000 tasks may open a subset of the 50,000 files, but the total number of file opens and task overhead grows. This is the opposite of what is needed to reduce per-task file opens.

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

Why it fails: Disabling broadcast joins has no effect on file layout or the number of files a scan opens. The scan still enumerates every small file in the Delta table regardless of join strategy. This setting only changes join execution and would not reduce the per-task file-open overhead caused by 50,000 small files.

Increase the number of executors to process more files in parallel.

Why it fails: While adding executors can increase parallelism, the small files problem is primarily due to per-file overhead in listing, opening, and reading metadata. Each file requires a separate task, and with 10,000 files, the scheduling and I/O overhead dominate. Adding executors may help slightly but does not address the root cause of excessive small files.

Convert the Parquet files to CSV format to reduce file size.

Why it fails: Converting to CSV would likely increase file size and lose schema and columnar benefits. CSV is not compressed by default and is slower to parse. This would worsen performance, not improve it. The issue is the number of files, not the format, and Parquet is already efficient for analytics.

Set spark.sql.files.maxPartitionBytes to a very large value to combine files into fewer partitions.

Why it fails: spark.sql.files.maxPartitionBytes controls the maximum size of a partition when reading files. Increasing it can reduce the number of partitions, but it does not eliminate the small files themselves. The overhead of listing and opening each file remains. It may help by grouping multiple small files into one partition, but it is not as effective as physically compacting the files.

Rely on Databricks Delta Cache by enabling it on the cluster.

Why it fails: Delta Cache accelerates reads by caching remote Parquet files on local SSD, but it does not eliminate the need to recompute the DataFrame lineage. Each action will still trigger a new scan of the Delta table unless the DataFrame is explicitly cached. Delta Cache reduces I/O latency but does not prevent repeated execution of the same transformation logic. It is a complementary optimization, not a replacement for caching the DataFrame.

Persist the base DataFrame with StorageLevel.MEMORY_AND_DISK after the first action.

Why it fails: Using MEMORY_AND_DISK is reasonable, but the scenario emphasizes avoiding recomputation of the full table scan. Persisting after the first action still requires the first action to fully scan the table, and subsequent actions will reuse the cached data only if the cache is materialized. More importantly, the question asks for the best way to avoid recomputing the full scan; a broadcast join or Delta caching would be more directly effective. Also, persisting a 200 GB DataFrame in memory is impractical with typical executor memory.

Call .cache() on the DataFrame after the first transformation.

Why it fails: Calling .cache() on a DataFrame stores partitions in memory with a default MEMORY_AND_DISK storage level, but it does not help here because the DataFrame is already being reused across actions. The issue is that the initial scan is recomputed for each action; caching the post-transformation DataFrame will not prevent the upstream Delta table scan from being re-executed if the cached DataFrame is not actually persisted before the next action. The driver memory size is also irrelevant to executor-side caching.

Enable adaptive query execution by setting spark.sql.adaptive.enabled to true.

Why it fails: Adaptive query execution (AQE) optimizes shuffle partitions and join strategies at runtime, but it does not prevent repeated reads of the same input partitions. AQE can improve aggregation performance by coalescing partitions, but the root cause here is likely the lack of caching, leading to recomputation. Enabling AQE is beneficial but not the most direct solution for repeated partition reads.

Increase the driver memory to 8 GB.

Why it fails: Driver memory is not the bottleneck when tasks repeatedly read the same partitions. The issue is executor-side I/O and recomputation. Increasing driver memory might help if the driver is overwhelmed with task scheduling or collecting results, but the symptom described points to redundant reads on executors, which caching would resolve. Driver memory tuning is irrelevant here.

Set spark.sql.shuffle.partitions to 2000.

Why it fails: Increasing shuffle partitions can improve parallelism for large aggregations, but it does not address the symptom of tasks repeatedly reading the same partitions. That symptom suggests data is not being cached or that the input is being re-read due to lack of persistence. Adjusting shuffle partitions may help with skew but is not the most direct fix for repeated reads of the same partitions.

Increase spark.executor.memory to 32g to give each executor more memory for the large partitions.

Why it fails: Adding executor memory may help avoid spills, but it does not change the fact that one task is processing a disproportionate amount of data. The task will still run much longer than others because it is handling a skewed key. Memory tuning alone will not redistribute the workload or fix the runtime imbalance.

Set spark.sql.autoBroadcastJoinThreshold=-1 to disable broadcasting and force a sort-merge join.

Why it fails: Disabling broadcast join affects join strategies, but this scenario is a groupBy aggregation, not a join. There is no broadcast to disable here. Even if there were, forcing a sort-merge join would not address skew in the grouping key. This setting is irrelevant to the reported long-running tasks.

Increase spark.sql.shuffle.partitions from 200 to 2000 to create more partitions.

Why it fails: Increasing the number of shuffle partitions may reduce the average partition size, but if a single customer_id has millions of rows, that key will still land in one partition. More partitions does not break up a single hot key. The long-running tasks would persist because the skew is at the key level, not the partition count level.

Call repartition(200) on the DataFrame immediately before the groupBy to evenly distribute the records.

Why it fails: A round-robin repartition redistributes rows evenly across partitions, but the subsequent groupBy triggers a shuffle that re-hashes rows by the grouping key, undoing the even distribution and recreating the skewed partition. It also adds an extra full shuffle, increasing runtime and I/O without fixing the root cause.

Increase spark.sql.shuffle.partitions from 200 to 10000 so every task receives fewer records.

Why it fails: Raising the shuffle partition count can reduce per-task record volume, but it does not guarantee that the skewed key is spread across partitions, since hash partitioning still routes all rows with the same key to one partition. The oversized partition can remain oversized, so the OOM may persist while also adding scheduler and small-file overhead.

Set spark.executor.memory to 64g and spark.executor.cores to 1 so each executor has maximum heap for the aggregation.

Why it fails: Adding heap space can delay the failure but does not change the fact that one task holds a disproportionately large partition. With data that skews heavily, even a very large executor heap may be insufficient, and reducing cores lowers parallelism. This treats the symptom rather than the partition-size imbalance.

Broadcast the 3 GB customers DataFrame with `broadcast(df_customers)` to eliminate the shuffle.

Why it fails: Broadcasting removes the shuffle for the small side, but the skew lives on the large transactions side, which is already the probe side. The single hot key's rows still flow into one task, so the straggler remains. A 3 GB broadcast also risks driver and executor memory pressure against the default broadcast threshold.

Call `df_transactions.coalesce(8)` before the join to reduce the number of input partitions.

Why it fails: Coalescing reduces partition count without a shuffle and is intended to shrink output files, not to redistribute keys. Fewer input partitions means fewer parallel tasks, so the skewed key's work is concentrated even further and the stage becomes slower, not faster.

Increase `spark.sql.shuffle.partitions` to 4000 so the hot partition is split across more tasks.

Why it fails: Raising the shuffle partition count spreads the overall key space across more reducers, but a single key is always routed to exactly one partition. The 35 percent hot key still lands in one task, so the long-running straggler persists while the other tasks simply get smaller, adding overhead without fixing the bottleneck.

Enable Kafka consumer group rebalancing by setting a unique group.id per executor.

Why it fails: Using a unique group.id per executor would cause each executor to consume all partitions independently, leading to duplicate data and increased load. This is not a valid scaling strategy and would worsen the backlog. The issue is not consumer group assignment but rather the processing capacity of the streaming job.

Set spark.streaming.kafka.maxRatePerPartition to a very high value to consume faster.

Why it fails: Increasing the maximum rate per partition would push more data into each batch, potentially overwhelming the processing engine further. The job is already falling behind; consuming faster without increasing processing capacity will exacerbate the backlog and increase shuffle spill. The goal is to balance consumption with processing, not to maximize intake.

Increase spark.sql.shuffle.partitions to match the number of Kafka partitions.

Why it fails: Matching shuffle partitions to Kafka partitions may help parallelism, but the growing batch duration and high shuffle spill indicate that each batch is accumulating more data than can be processed in the trigger interval. The root cause is likely that the processing rate is lower than the input rate, causing a backlog. Simply changing partition count does not address the fundamental throughput mismatch.