Select Two (Multi-Select) Questions

spark.open('table_name')

Why it fails: There is no 'spark.open()' method in the SparkSession API. The reader API uses 'read', and the catalog API uses 'table'. Inventing methods that do not exist is a common trap; it is important to stick to the documented 'read' or 'table' methods provided by the Spark library.

spark.from_table('table_name')

Why it fails: The 'from_table' method does not exist in the SparkSession or DataFrameReader classes. Developers must use the standard 'table' or 'read.table' methods. Using incorrect method names will result in an AttributeError, as these functions are not part of the public Spark developer API for DataFrame creation.

Standard SQL Databases (JDBC)

Why it fails: While you can read from JDBC sources using batch queries, they are not natively supported as streaming sources because they lack a built-in offset tracking mechanism that Spark needs. Using them in a streaming context requires workarounds like foreachBatch and custom logic, not native stream reading.

In-memory lists

Why it fails: In-memory collections (like Python lists or Scala sequences) are not supported as streaming sources. Structured Streaming expects an external, persistent source that provides a stream of data with identifiable offsets, which in-memory data structures cannot provide for a continuous and recoverable streaming pipeline.

The client maintains a local SparkContext that runs tasks in parallel with the remote server to reduce latency.

Why it fails: Spark Connect clients do not create or maintain a local SparkContext. The architecture intentionally removes the driver from the client. All task scheduling and execution occur on the server. The client only builds logical plans and sends them over gRPC, so there is no local task parallelism to reduce latency.

User-defined functions defined with @udf are executed in the client Python process to avoid shipping code to the server.

Why it fails: In Spark Connect, Python UDFs are serialized and sent to the server, where they are executed in Python worker processes alongside the Spark executors. They are not executed in the client process. This ensures the UDF runs where the data resides, but it also means the UDF code must be serializable and available to the server environment.

df.show() executes entirely on the client by sampling data from a local cache maintained by Spark Connect.

Why it fails: Spark Connect does not maintain a local data cache on the client for show(). The show() action sends the plan to the server, which computes and returns a formatted preview. The client only receives and displays the small result. There is no client-side cache of DataFrame contents, so the operation depends on the remote server.

`psdf.to_pandas()`

Why it fails: `to_pandas()` collects the entire DataFrame to the driver, which is highly inefficient for large datasets and can cause OutOfMemoryError. It is the opposite of a distributed operation. This is not an efficient operation for large data and should be avoided unless the dataset is small enough to fit in driver memory.

`psdf.head(20)`

Why it fails: `head(20)` collects the first 20 rows to the driver as a pandas DataFrame. While it is a small collection, it still brings data to the driver and is not a purely distributed operation. It is efficient in terms of data volume but does not keep the result distributed. Therefore, it is not considered an efficient distributed operation in the same sense as aggregations or joins.

`psdf['value'].apply(lambda x: x * 2)`

Why it fails: Using `apply` with a Python lambda on a Series forces row-by-row execution, which is slow and may collect data to the driver or cause serialization overhead. In Pandas API on Spark, `apply` is not vectorized and can lead to performance degradation. It is not considered an efficient distributed operation compared to built-in functions.

Executors are responsible for scheduling tasks across the worker nodes.

Why it fails: Task scheduling is performed by the Driver, which determines the optimal distribution of work based on data locality and available resources. Executors do not perform scheduling; they merely receive task definitions from the Driver and execute the serialized code provided within those tasks on their assigned compute slots.

Executors create the physical execution plan from user code.

Why it fails: The creation of the physical execution plan is a core responsibility of the Catalyst Optimizer and the DAG Scheduler residing within the Driver. Executors have no knowledge of the overall query plan or the logical structure of the application; they only focus on completing individual units of work.

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

Why it fails: Resource allocation is managed by the Cluster Manager, which negotiates with the underlying resource manager to provide containers for executors. Executors are passive recipients of these resources and do not participate in the negotiation or allocation process, as they are managed entities within the allocated container space.

ARRAY_MAX(tags)

Why it fails: ARRAY_MAX returns the maximum value in an array, which for strings would be the lexicographically largest element. It does not check for a specific value like 'urgent' nor count elements. This function is useful for numeric arrays to find the peak value, but it is irrelevant to the requirements of filtering by membership or counting array length.

EXPLODE(tags)

Why it fails: EXPLODE is a generator function that transforms an array column into multiple rows, one per element. While it can be used to find arrays containing a value via a filter, it changes the row granularity and requires aggregation to restore the original rows. It is not suitable for simply filtering rows or producing a count column. Using EXPLODE here would complicate the query unnecessarily.

COLLECT_LIST(tags)

Why it fails: COLLECT_LIST is an aggregate function that gathers values into an array, typically after a GROUP BY. It does not test for membership or count elements within an existing array. Applying it to an array column would create nested arrays and is not appropriate for filtering or counting. It serves a different purpose: aggregating rows, not inspecting array contents.

Increase spark.sql.shuffle.partitions.

Why it fails: Increasing shuffle partitions actually tends to increase the number of small files and metadata operations. While it might prevent OOM errors, it does not reduce the volume of data shuffled. Instead, it changes how that data is split, often leading to higher overhead due to more task scheduling.

Enable dynamic resource allocation.

Why it fails: Dynamic allocation helps manage cluster resources by adding or removing executors based on workload. While it improves overall cluster efficiency and cost, it does not optimize the data flow within a single Spark job or reduce the actual volume of data transferred during shuffle operations.

Increase spark.driver.memory.

Why it fails: Driver memory is independent of the shuffle process performed by executors. While more driver memory is required for large result sets being collected, it has no impact on the volume of data being shuffled between executors during distributed operations like joins or group-by transformations.

Executors use fixed memory boundaries that cannot be adjusted during task execution.

Why it fails: Spark uses a unified memory manager that allows dynamic borrowing between storage and execution regions. This flexibility prevents rigid memory boundaries, ensuring that memory is not wasted if one region is underutilized while the other requires more capacity for processing or caching operations.

Executors are allowed to access the Driver's memory pool to store large datasets.

Why it fails: The Driver's memory is isolated and reserved for metadata, task scheduling, and collected results. Executors cannot access the Driver's memory pool. If an executor runs out of memory, it must spill data to local disk or fail, as it has no capability to offload tasks to the Driver.

Memory management is handled entirely by the Cluster Manager, not the Spark process.

Why it fails: While the Cluster Manager allocates the total memory per node, Spark's internal memory manager (MemoryManager) sub-divides this space into specific regions. The Cluster Manager has no visibility into how Spark partitions its memory between execution and storage pools; that logic is implemented solely within the Spark application.

The total amount of disk space on the worker nodes.

Why it fails: While disk space is necessary for overflow or shuffle files, it is not a direct factor in determining the degree of parallelism. Parallelism is defined by compute resources, not storage capacity; having large disks does not allow more tasks to run in parallel on an executor.

The network latency between the driver and the cluster manager.

Why it fails: Network latency between the driver and the cluster manager impacts the time taken for negotiation and initial startup, but it does not dictate the parallelism of the tasks themselves once they are running on the executors, which is determined by data and compute limits.

The size of the broadcast variables.

Why it fails: Broadcast variables are used to distribute read-only data efficiently to all nodes. While their size affects memory usage and network overhead during the broadcast phase, they do not inherently limit or dictate the degree of task parallelism within the cluster for the job's execution.

Manually restart the cluster every time a query takes more than 10 seconds.

Why it fails: Restarting a cluster is a destructive and time-consuming operation that does not address the underlying query inefficiency. It is not a debugging tool. Instead, developers should profile the code to understand the performance bottleneck and optimize the query itself, rather than trying to brute-force a solution through cluster restarts.

Increase the driver memory to the maximum available for all jobs.

Why it fails: Increasing driver memory is a reactive measure for Out-Of-Memory errors on the driver, not a performance debugging tool. It does not provide insight into why a query is slow. Over-allocating memory can lead to resource waste and does not resolve issues related to inefficient SQL logic or data skew.

Increase spark.sql.shuffle.partitions to a much larger value without changing the query plan.

Why it fails: Raising the shuffle partition count creates more, smaller output files per task but does not reduce the total bytes shuffled. The same records still cross the network, so shuffle write volume stays essentially the same while adding scheduling overhead and more small files.

Call persist(MEMORY_ONLY) on the DataFrame before the shuffle stage.

Why it fails: Caching a DataFrame in memory speeds up repeated access but does not alter the number of records repartitioned during a shuffle. The shuffle still writes the same data to disk, so caching does not address the root cause of the large shuffle write volume described.

Enable the Kryo serializer instead of the default Java serializer for the shuffle.

Why it fails: Kryo reduces the serialized size of each record, which can shrink shuffle bytes somewhat, but it does not change how many records are shuffled. The question asks for actions most likely to reduce shuffle volume structurally, and serialization tuning is a secondary, smaller-effect optimization compared to eliminating or pre-aggregating the shuffle.

The DAG Scheduler ensures that broadcast variables are only used in narrow transformations.

Why it fails: The DAG Scheduler creates stages from the logical DAG but does not enforce constraints on where broadcast variables can be used. Broadcast variables can be used in both narrow and wide transformations. The DAG Scheduler's responsibility is stage partitioning, not variable usage restrictions.

The Cluster Manager replicates the broadcast variable across all nodes in the cluster.

Why it fails: The Cluster Manager is not involved in distributing broadcast variables. Its role is to allocate resources like executors to the Spark application. Broadcast variables are handled entirely by the Spark driver and executors, independent of the Cluster Manager's resource allocation duties.

The Task Scheduler assigns the broadcast variable to each task individually.

Why it fails: The Task Scheduler schedules tasks onto executors but does not handle the distribution of broadcast variables. Broadcast variables are distributed once per executor, not per task. The Task Scheduler's role is to manage task execution, not data distribution.

Use `df.orderBy('event_time').rdd.zipWithIndex()` and manually look up the previous index per user in a driver-side dictionary.

Why it fails: This fails because `zipWithIndex` assigns a global index after a global sort, not a per-user sequence, so the 'previous index' does not correspond to the user's prior event. Additionally, collecting a driver-side dictionary scales poorly and defeats the purpose of distributed Spark processing for a large `events` table.

Apply `window('user_id', 'event_time')` with `lag('payload')` and rely on Spark to fill nulls for the first event.

Why it fails: This is invalid because `window` is not a function that accepts column names; the correct API is `Window.partitionBy('user_id').orderBy('event_time')`. Also, the frame specification matters: default lag still returns null for the first row, but the syntax as written would not even compile, so it cannot satisfy the scenario.

Call `groupBy('user_id').agg(collect_list('payload'))` and then explode the resulting list back into rows.

Why it fails: This fails because `groupBy` plus `collect_list` collapses rows per user and loses the original row-level ordering unless you also sort inside the aggregation, which is not guaranteed. Exploding the list also loses the association with the original `event_time` values, so the enriched DataFrame would not correctly align payloads to their preceding events.

A shuffle eliminates the need for the driver to coordinate task scheduling across stages.

Why it fails: This is incorrect. The driver, specifically the DAG Scheduler, still coordinates stage execution and task scheduling. In fact, shuffles introduce additional coordination because the DAG Scheduler must wait for map stages to complete before submitting reduce stages. The shuffle service on executors handles data transfer, but the driver orchestrates the overall job flow, including stage retries if shuffle outputs are lost.

A shuffle always results in exactly one partition per reducer task, regardless of the number of map tasks.

Why it fails: This is incorrect. The number of reduce partitions is determined by the `spark.sql.shuffle.partitions` (for SQL/DataFrame) or the `numPartitions` parameter of the transformation (e.g., `groupByKey(numPartitions)`). It does not automatically match the number of map tasks. The number of map tasks (and thus map output files) can be different, and each reducer fetches only its relevant partitions from all map outputs.

A shuffle is only triggered by actions, not by transformations.

Why it fails: This is incorrect. Shuffles are triggered by certain transformations that require data redistribution, such as `groupByKey`, `reduceByKey`, `join`, and `repartition`. Actions like `count` or `collect` trigger job execution but do not directly cause shuffles; they simply force the evaluation of the DAG, which may include shuffle transformations. Thus, shuffles are a property of transformations, not actions.

Use an `INNER JOIN` instead of an outer join.

Why it fails: An `INNER JOIN` returns only rows where `customer_id` matches in both tables, discarding transactions with no customer. That directly contradicts the requirement to keep unmatched transactions with null customer columns. While it avoids some null handling, it fails the primary retention condition, making it unsuitable regardless of deduplication.

Use a `FULL OUTER JOIN` between the two tables.

Why it fails: A `FULL OUTER JOIN` returns unmatched rows from both sides, which would include customers with no transactions. The scenario only requires retaining unmatched transactions, so this introduces extra rows the engineer did not ask for. It also does not by itself eliminate duplicates, so it does not satisfy either stated condition cleanly.

Add a `GROUP BY` on all columns from both tables.

Why it fails: Grouping by every column effectively deduplicates, but it is a heavy and awkward substitute for `DISTINCT` and can interfere with null handling in outer joins. It also changes the query's semantics and may not produce the intended row-level output. It does not address the left-outer retention requirement and is not the idiomatic deduplication tool here.

Set `spark.sql.adaptive.enabled` to false to stabilize the connection.

Why it fails: Adaptive query execution changes runtime plan behavior on the server but has no bearing on client-server transport reliability. Toggling it does not prevent gRPC disconnects or message-size failures and may degrade performance. This setting is unrelated to connectivity, so it is not a valid practice here.

Increase `spark.sql.shuffle.partitions` to 2000 to reduce the number of RPC calls.

Why it fails: Shuffle partition count governs how many partitions a shuffle produces; it has no effect on the number or size of gRPC calls between client and server. Increasing it can actually add overhead and does not address connection errors. The setting is unrelated to transport stability, so it is not a valid practice for this scenario.

Disable TLS between the client and the cluster to reduce handshake overhead.

Why it fails: Disabling TLS weakens security and is not supported for Databricks Spark Connect connections, which require encrypted transport. It does not reliably fix intermittent connection errors and would violate security requirements. Reducing handshake overhead is not a legitimate troubleshooting step for stability, so this option is incorrect.

COUNT(customer_id) FILTER (WHERE customer_id IS NOT NULL)

Why it fails: This expression counts all non-null customer_id rows per region, including duplicates. It does not deduplicate customers who appear in multiple orders, so a region with repeated purchases by the same customer is overcounted. The FILTER clause only removes nulls; it does not make the count distinct, so it fails the requirement of counting unique customers.

SUM(DISTINCT customer_id)

Why it fails: SUM(DISTINCT customer_id) adds up unique customer_id values rather than counting them, producing a numeric total that has no meaning for a count of customers. It is also only valid for numeric columns and would fail or mislead if customer_id were a string. It does not answer how many distinct customers exist per region.

COLLECT_SET(customer_id)

Why it fails: COLLECT_SET returns an array of distinct customer_id values per region, not a count. While the array size could be computed with SIZE, COLLECT_SET alone yields a collection that is expensive to materialize for high-cardinality groups and is not a count expression. It does not directly produce the numeric distinct customer count required in the result set.

Use the MERGE command to combine the tables.

Why it fails: MERGE is used for upserts into a Delta table, not for joining two tables for analysis. It is not a join operation and would not produce a joined result set. It is used for transactional writes, not for query optimization. Therefore, it is irrelevant to the scenario.

Partition the customers table by customer_id to match the orders table.

Why it fails: Partitioning the customers table by customer_id would not help the join because the orders table is partitioned by order_date, not customer_id. Partitioning by customer_id would create many small partitions and might not align with the join key distribution. Moreover, partitioning a large table by a high-cardinality column like customer_id is generally not recommended due to small file issues.

Repartition both tables by customer_id before the join.

Why it fails: Repartitioning both tables by customer_id would ensure that matching keys are in the same partition, but it incurs a full shuffle of both large tables. This is expensive and not the most efficient approach when one table can be broadcast. It might be necessary if both tables are large, but the question asks for actions to minimize shuffling, and broadcasting is better when applicable.

Set the output mode to `complete`.

Why it fails: Output mode controls how results are written, not the delivery guarantee. `Complete` mode rewrites the entire result table each time, which is inefficient and not related to exactly-once semantics. Exactly-once is achieved through checkpointing and idempotent sinks. Therefore this option does not contribute to the requirement and is incorrect.

Use the `foreachBatch` sink to write to the Delta table.

Why it fails: While `foreachBatch` can be used for custom logic, it does not by itself guarantee exactly-once semantics. The sink must be idempotent or transactional. Delta Lake's native sink already provides exactly-once when used with a checkpoint. Using `foreachBatch` adds complexity and requires manual handling of idempotency, so it is not a required option for exactly-once in this scenario.

Use `trigger(processingTime='0 seconds')` for continuous processing.

Why it fails: The trigger interval controls how often micro-batches are executed, not the delivery guarantee. Continuous processing mode is experimental and does not support all operations. It does not provide exactly-once by itself. Therefore this option is not required and does not help achieve the stated goal. It is a distractor related to performance tuning rather than fault tolerance.

Increase spark.sql.shuffle.partitions to a much larger number so each task handles fewer rows.

Why it fails: The symptom is that many output partitions are empty or nearly empty downstream, not that each task is too large. Raising the partition count would create even more tiny tasks and more scheduling overhead. The real issue is the volume of data entering the shuffle, which partitioning cannot reduce. This change would likely make the job slower.

Repartition the DataFrame on the join key before the shuffle to colocate matching rows.

Why it fails: An explicit repartition on the join key introduces an extra full shuffle before the join's own Exchange, doubling network and disk cost. Spark already partitions on the join key during the Exchange, so this is redundant. It does not reduce the amount of data shuffled and adds a stage. This would worsen rather than improve the runtime.

Cache the shuffled DataFrame with persist so downstream stages reuse the same partitions.

Why it fails: Caching helps when the same DataFrame is reused across multiple actions, but the problem described is wasted shuffle volume in a single lineage, not repeated reads. Persisting the post-shuffle data would hold large partitions in memory with no reuse to amortize the cost. It also risks eviction and recomputation. Caching is not the right remedy for excessive shuffle input.