Free Databricks Certified Associate Developer for Apache Spark Sample Questions

Correct answer: A — Three separate jobs, because Spark launches one job per action, and each job is then broken into stages and tasks that run on the executors.

• A. Correct. Each action triggers Spark's lazy-evaluated transformations to actually execute, and the driver submits a distinct job per action, which is then split into stages and tasks.
• B. Incorrect. Spark does not defer job submission to the end of the script; each action independently triggers its own job as soon as it is called.
• C. Incorrect. `count()` still requires reading and aggregating the underlying partitions across executors; it is not resolved purely inside the driver without executor work.
• D. Incorrect. Each action produces its own job in the Spark UI; `collect()` and `count()` are not merged into a job triggered by an unrelated `write.parquet` call on a different DataFrame.
• E. Incorrect. Job counting is tied to actions, not to shared lineage or a shared SparkSession; three actions on three DataFrames still produce three jobs even under one session.

Correct answer: B — Cluster mode

• A. This is incorrect because in this mode the driver runs on the machine that issued the submit command, so the application would stop if that machine disconnected.
• B. This is correct because this deploy mode launches the driver process on a node inside the cluster itself, so the application continues running independently of the machine that originally issued the submit command.
• C. This is incorrect because this mode runs the driver and executors within a single JVM on one machine and is not used for submitting a job to a YARN cluster.
• D. This is incorrect because this term refers to Spark's built-in cluster manager, not to a deploy mode that controls where the driver process runs.
• E. This is incorrect because this is not a recognized Spark deploy mode; the deploy mode options control driver placement, not an interactive session type.

Correct answers: A, B — `total = df3.count()`, because `count()` is an action that forces Spark to execute the accumulated transformations and return a value to the driver.; `df3.write.mode("overwrite").parquet("/data/2026")`, because writing output is an action that forces Spark to execute the transformation chain and materialize files.

• A. This is correct: `count()` is a classic action that returns a value to the driver, forcing Spark to execute the read, column addition, and filter that precede it in the lazy plan.
• B. This is also correct: writing to storage is an action, so this line triggers its own job that executes the full transformation chain again to produce and persist the output files.
• C. `spark.read.parquet()` only builds a logical plan describing the source; it does not eagerly load records into driver memory, so this line alone does not trigger a job.
• D. `withColumn()` is a transformation, not an action; it lazily extends the logical plan with a new column expression without materializing any values at this point.
• E. `filter()` is a transformation, not an action; row filtering is recorded in the plan and only executes once a downstream action like `count()` or `write()` runs.

Correct answers: A, B — `coalesce()` can only decrease a DataFrame's partition count and can never raise it beyond however many partitions already exist.; `coalesce()` avoids a full shuffle by combining existing partitions on the same executor, moving less data than `repartition()`.

• A. Coalesce is designed only to merge existing partitions together, so it cannot raise the partition count above what the DataFrame already has; requesting more partitions than currently exist has no effect.
• B. By combining nearby partitions on the same executor instead of redistributing every row across the network, coalesce moves far less data than a full shuffle, which is exactly why it is the cheaper way to reduce partition count.
• C. Because coalesce merges whatever partitions already exist rather than hashing rows into new ones, the resulting partitions can be uneven in size, unlike the more even distribution a full shuffle in `repartition` typically produces.
• D. A full shuffle moving every row across the cluster is the behavior of `repartition`, not `coalesce`; coalesce specifically avoids that cost when it is only reducing the partition count.
• E. Coalesce only changes how a DataFrame's existing partitions are grouped together; it has no effect on the DataFrame's schema and does not drop or recompute any columns.

Correct answer: A — Spark exposes a unified engine where Spark SQL, DataFrames, Structured Streaming, and MLlib share the same execution model, so ETL, reporting, and forecasting run on one cluster.

• A. This is correct: Spark's unified engine lets the same cluster and APIs handle batch ETL, SQL reporting, and MLlib model training, which is exactly the consolidation the company wants. This eliminates the need to operate three separate specialized systems.
• B. Spark does not translate jobs into MapReduce plans; it compiles work into its own DAG of stages and tasks executed by its own scheduler. Framing Spark as a MapReduce-compatibility layer misstates how its execution engine works.
• C. Spark does not enforce schema-on-write, and reading raw files does not remove the need for a query engine to serve reporting workloads. Schema inference and enforcement are configurable, not automatic replacements for a SQL layer.
• D. Spark itself has no built-in charting dashboard; visualization is handled by separate tools such as notebooks or BI products that connect to Spark. This option describes a capability Spark does not provide.
• E. Spark does not auto-provision GPU pools when ML code runs; hardware allocation depends on the cluster manager and the cluster configuration chosen by an administrator. This overstates what happens automatically.

Correct answer: A — No Spark job is submitted to the cluster; Spark only records `filtered` and `grouped` as logical plans built on top of `df`, since every call in the cell is a transformation.

• A. This is correct: `read`, `filter`, and `groupBy`/`sum` are all transformations, so the cell only builds up a logical plan referencing `df`. No job is submitted to the cluster because no action was called.
• B. This is incorrect because no action appears anywhere in the cell, so no job is submitted at all, let alone two separate ones. Filtering and grouping stay purely as plan-building steps until something forces execution.
• C. This is incorrect because being a wide transformation does not make `groupBy` eager. Wide transformations still only add to the logical plan and only cause a shuffle when a later action forces the plan to run.
• D. This is incorrect because leaving a DataFrame unused by an action is not an error condition in Spark; the cell simply finishes with `grouped` holding an unexecuted plan, and no exception is raised.
• E. This is incorrect because Spark does not selectively execute some transformations eagerly based on file metadata. Every transformation in the chain, including the filter, remains unexecuted until an action is called.

Correct answer: A — `pipeline = Pipeline(stages=[indexer, lr])` `model = pipeline.fit(train_df)`, which chains both stages in order and trains them together with one `fit()` call.

• A. `Pipeline` accepts an ordered list of stages, and a single `fit()` call runs each stage's transformation or training in sequence, which is exactly the reusable, chained workflow the developer wants.
• B. `indexer.fit(train_df)` produces a fitted `StringIndexerModel`, not a transformed DataFrame; `lr.fit()` needs an actual DataFrame of features and labels, so feeding it a model object is a type mismatch.
• C. A Python list is not a Spark ML construct and has no `.fit()` method, so this line fails immediately rather than chaining the indexer and the classifier.
• D. `LogisticRegression.fit()` only accepts a DataFrame (and optional params), not a `transformers` argument, so this call does not exist as written and raises a `TypeError`.
• E. `Pipeline` expects its stages to be passed as a `stages` keyword argument holding a list, not as separate positional arguments, so constructing it this way raises a `TypeError` before fitting begins.

Correct answer: E — `batch_df.write.format("delta").mode("append").save(path)`, which commits the new rows as an additional atomic transaction on top of the existing table history

• A. Overwrite mode replaces the entire table contents with the new batch, so each hourly run would erase every transaction written in earlier runs rather than accumulating them.
• B. Errorifexists mode is the default guard against accidentally clobbering an existing table and raises an exception once the path already holds data, so it cannot support repeated hourly writes at all.
• C. Spark's DataFrameWriter has no automatic key-based merge behavior triggered by omitting the mode; without an explicit mode it falls back to `errorifexists`, which would fail after the first successful write.
• D. Ignore mode is designed to silently skip a write only when the target already contains data, so it would drop every batch after the first hour instead of accumulating the streaming transactions.
• E. Append mode commits each new batch as an additional atomic transaction without touching existing rows, so every hour's transactions accumulate on top of the prior history exactly as the job requires.

Correct answer: A — `createTempView` raises an exception if a temporary view with that name already exists, while `createOrReplaceTempView` silently overwrites it.

• A. This is the documented behavioral distinction: the non-replacing method fails with an already-exists error on a duplicate name, while the replacing method overwrites the prior registration without error.
• B. Both methods register a session-scoped temporary view in the session's default namespace; neither one uses the `global_temp` database, which is reserved for the separate global temp view methods.
• C. Neither method writes anything to the metastore; both create purely in-memory, session-scoped catalog entries that are lost when the session ends.
• D. Neither method requires the DataFrame to be cached beforehand; registering a view only binds a name to the DataFrame's logical plan, independent of caching.
• E. Both methods take only the view name string as their argument; neither accepts a SQL query string or a column list.

Correct answers: A, B — Calling `df.createOrReplaceTempView("orders")` in the first notebook will not make `orders` visible to a query in the second notebook, because a plain temp view is scoped to the session that created it.; Calling `df.createOrReplaceGlobalTempView("orders")` in the first notebook makes the view visible from the second one too, because global temp views are shared across sessions in the same Spark application.

• A. Correct — a session-scoped temp view created with `createOrReplaceTempView` only exists for the Spark session that created it, so a different session's queries cannot see it by name.
• B. Correct — a global temp view is tied to the `global_temp` system database, which is shared by every session running in the same Spark application, so other sessions can query it.
• C. Incorrect — the `global_temp.` prefix is required for every session accessing a global temp view, regardless of whether the query runs from a SQL cell, `spark.sql`, or another language.
• D. Incorrect — a plain temp view is an in-memory, session-local logical plan; it is never written to the metastore, so no other session or later job can discover it as a table.
• E. Incorrect — global temp views live only in memory for the lifetime of the Spark application; they are not persisted to disk and disappear once the application (and its cluster) stops.

Correct answer: A — transactions_df.write.mode("overwrite").partitionBy("transaction_date").saveAsTable("transactions")

• A. `partitionBy` writes the table as Hive-style directories keyed by `transaction_date`, so a later query filtering on that column only reads the matching directory. This is the standard way to get file-skipping without introducing bucketing or sort metadata.
• B. `bucketBy` hashes rows into a fixed number of bucket files rather than creating per-value directories, so a reader cannot tell from the file layout which bucket holds a given `transaction_date`. It optimizes joins and aggregations on the bucketing column, not date-based file skipping.
• C. `sortBy` can only be used together with a preceding `bucketBy` call on the same writer. Calling it alone raises an error before any data is written, so this call never reaches the metastore.
• D. `DataFrameWriter` has no `orderBy` method; ordering a DataFrame before a write is done with `DataFrame.orderBy`, not on the writer object. This call fails with an attribute error at runtime.
• E. `repartition` changes how many in-memory Spark partitions are used during the write, which can affect the number of output files, but it does not create the on-disk directory structure that lets a reader prune by `transaction_date` value.

Correct answer: C — withColumn

• A. select projects a set of column expressions but does not take a target column name and a value expression as two positional arguments the way this call is written, so this signature does not fit.
• B. withColumnRenamed only accepts an existing column name and a new name string; it cannot evaluate a when/otherwise expression, so it cannot produce a computed column.
• C. withColumn takes a target column name and a Column expression, adding the result as a new column while returning a new DataFrame, which matches this call exactly.
• D. filter restricts which rows are kept based on a boolean condition; it does not add or compute a new column, so it does not fit this signature.
• E. withColumnsRenamed takes a dictionary mapping existing names to new names for bulk renaming, not a name plus a computed expression, so it does not match this call.

Correct answer: A — sensor_df.filter(F.isnan("reading"))

• A. isnan() specifically flags Spark's floating-point NaN marker used for invalid numeric results, distinguishing it from a genuinely missing None value.
• B. isNull() matches rows where the value is Python None/SQL NULL, which is exactly the missing-data case this scenario wants to exclude, not the invalid calculation.
• C. Filtering with isNotNull() returns both valid numeric readings and NaN rows together, since NaN is a defined floating-point value rather than a null, so it fails to isolate the invalid rows.
• D. na.drop(subset=["reading"]) removes rows with null values and returns everything else, which keeps the NaN rows mixed in with valid readings rather than isolating them.
• E. isnull() is functionally the same null check as isNull(), so it identifies missing-data rows rather than the NaN rows produced by the invalid calculation.

Correct answer: C — `orders.agg(F.count("*").alias("total_orders"), F.avg("order_total").alias("avg_total"))`, which applies both aggregate expressions across the whole DataFrame and returns one row labeled by the given aliases.

• A. Once `.select()` has collapsed the data to a single aggregated row, grouping by `order_total` afterward just partitions that one row by its own value, which does not restore a meaningful per-order breakdown or match the requested single summary row.
• B. `.alias()` on a DataFrame does not rename a column or introduce an average; it is a no-op renaming of the DataFrame reference itself, so no average column is produced.
• C. This is correct: calling `.agg()` directly on the DataFrame without a preceding `.groupBy()` key aggregates across every row, and the two aliased expressions produce exactly the two labeled columns the report needs in one row.
• D. Writing two comma-separated expressions on one line creates a tuple of two separate results in Python rather than combining them into a single DataFrame row, so it does not produce the requested report.
• E. `DataFrame` objects do not expose a bare `.avg()` method, and adding an aggregate function as a new column outside of `.agg()` or `.select()` is not valid, so this raises an error before producing any result.

Correct answer: C — `null`, because `promotions` has no row with `order_id` = 102, so the left join keeps the order row with a null for the unmatched column.

• A. Spark does not substitute empty strings for missing values produced by a join; unmatched right-side columns are populated with `null`, not `""`.
• B. A left join keeps every row from the left DataFrame no matter what, filling right-side columns with null when there is no match, rather than dropping the row.
• C. This is correct. A left join preserves all rows from `orders`, and since `promotions` has no row for `order_id` 102, the `code` column for that row is populated with `null`.
• D. Spark does not borrow a value from an unrelated matched row; there is no fallback to reuse `"SAVE10"` for an order that has no corresponding promotion row.
• E. Spark never copies a value from a different column, such as `item`, into an unrelated output column; a missing match produces a null, not a reused string from elsewhere in the row.

Correct answer: A — Swap the arguments to `datediff(current_date(), col("signup_date"))`; `datediff(end, start)` returns `end` minus `start` in days, so the earlier date first yields a negative count.

• A. `datediff()` computes its first argument minus its second argument in days, so passing the earlier `signup_date` as the first argument and today as the second produces a negative number; swapping the order fixes the sign.
• B. `date_add()` shifts a single date by a fixed number of days and does not compute the elapsed days between two columns at all, so it cannot replace `datediff()` for this calculation.
• C. `datediff()` works correctly with `DateType` arguments and the column's data type is not the source of the negative sign; the cause is the order in which the two dates are passed.
• D. Wrapping the result in `abs()` hides the sign without fixing the underlying argument order, and the claim that `datediff()` always returns a negative value for an earlier first argument describes exactly the bug rather than a real universal rule to rely on.
• E. `months_between()` returns a number of months, not days, and `datediff()` returns days, not months, so this option reverses what each function actually measures.

Correct answer: B — schema(order_schema)

• A. Turning on `inferSchema` makes Spark scan the file to guess column types, which is exactly the slow, unreliable path the engineer needs to avoid, and it would not guarantee `order_id` comes back as a long.
• B. Passing the `StructType` directly forces every row to be parsed according to the declared field names and types, so `order_id` is read as a `LongType` with no inference scan required.
• C. Declaring a header only tells Spark to treat the first line as column names; it says nothing about column types and does nothing to prevent inference from running.
• D. Chaining `format` and `load` is just an alternate way to specify the CSV source path and does not attach a schema, so the file would still fall back to default typing behavior.

Correct answer: A — 3, 2, 1

• A. `ascending=False` reverses the sort direction for the single `id` column, so rows are ordered from the highest value to the lowest, producing 3, then 2, then 1.
• B. This would be the result of an ascending sort, but `ascending=False` explicitly requests descending order, so the lowest-to-highest sequence 1, 2, 3 is not what this call produces.
• C. `orderBy()` always reshuffles rows into the requested sort order; a `DataFrame` has no guaranteed row ordering to preserve in the first place, so the output reflects the sort, not the construction order.
• D. `ascending` accepts a single boolean when sorting by one column; a list is only required when sorting by multiple columns and specifying a direction per column, so a bare `False` here is valid and does not raise an error.
• E. This sequence matches neither a full ascending nor a full descending sort on `id`, so it does not correspond to any ordering that `orderBy("id", ascending=False)` would produce.

Correct answer: A — `spark.sql.autoBroadcastJoinThreshold` sets the largest estimated table size Spark broadcasts automatically, defaulting to 10 MB unless the session overrides it.

• A. This is correct: `spark.sql.autoBroadcastJoinThreshold` is the size cutoff the planner compares a table's estimated size against, and its default of 10 MB is why small dimension tables get broadcast without any explicit hint. Setting it to -1 disables this automatic behavior entirely.
• B. `spark.sql.shuffle.partitions` only controls the number of partitions used after a shuffle occurs and has nothing to do with the size threshold that triggers automatic broadcasting. Its default of 200 is unrelated to table size estimation.
• C. `spark.sql.broadcastTimeout` governs how long a broadcast exchange is allowed to run before timing out, not what size table qualifies for broadcasting in the first place. Its 300-second default is a wait limit, not a size limit.
• D. `spark.driver.maxResultSize` limits how much data a driver can collect from actions like `collect()`, and it is not the property the planner consults when deciding whether a table is small enough to broadcast. Raising it does not change broadcast eligibility.
• E. Automatic size-based broadcasting is independent of `spark.sql.adaptive.enabled` and works whether or not adaptive query execution is turned on. AQE can additionally convert a join to broadcast at runtime, but it is not a prerequisite for the size-based default behavior.

Correct answer: A — Wrap `country_codes` with `spark.sparkContext.broadcast(country_codes)` and reference `bc.value.get(code, "Unknown")`, so every executor keeps one cached copy.

• A. Broadcasting caches a single read-only copy of the dictionary on each executor, and referencing `.value` inside the UDF avoids re-serializing and re-sending the dictionary with every task.
• B. Rebuilding the dictionary inside the function makes the problem worse: Spark still ships whatever data built it, and the work of constructing the mapping now repeats on every single row.
• C. A join is a valid pattern for larger reference tables, but for a small in-memory dictionary it adds an unnecessary shuffle and does not fix the serialization behavior of the existing UDF code.
• D. Accumulators are write-only aggregators whose running total only the driver can read; they are not a mechanism for distributing an arbitrary read-only mapping for executors to look values up in.
• E. `spark.sql.autoBroadcastJoinThreshold` only influences whether Catalyst chooses a broadcast join strategy for DataFrame joins, and has no effect on a plain Python closure referenced inside a UDF.

Correct answer: C — GroupStateTimeout.ProcessingTimeTimeout()

• A. This disables automatic expiration entirely, so state for a group is retained indefinitely until the per-group function itself calls `state.remove()`, which does not satisfy a fixed wall-clock expiration requirement.
• B. This ties expiration to the watermark computed from an event-time column in the data, so how quickly state expires depends on data arrival patterns rather than a fixed span of processing time.
• C. This measures elapsed wall-clock time against the duration set with `state.setTimeoutDuration()`, independent of any event-time column or watermark, which is exactly the behavior described.
• D. `GroupStateTimeout` has no member with this name; the only valid timeout configurations are `NoTimeout()`, `ProcessingTimeTimeout()`, and `EventTimeTimeout()`.

Correct answers: C, D — Every call to `state.update()` inside the per-group function must pass a two-element tuple whose values line up with `count` and `last_seen`, since the value is validated against `stateStructType`.; Because `timeoutConf` is `GroupStateTimeout.NoTimeout()`, `state.hasTimedOut` will always evaluate to `False` inside the function, so no branch of the function will ever run purely due to a timeout.

• A. Calling `state.setTimeoutDuration()` while `timeoutConf` is `NoTimeout()` raises an error at runtime, since duration-based timeouts are only valid when `ProcessingTimeTimeout()` is the configured timeout type.
• B. State declared with `stateStructType` is maintained independently per grouping key produced by `groupBy()`; each distinct key gets its own isolated state entry, not one shared object across keys.
• C. `state.update()` validates its argument against the declared `stateStructType`, so passing a tuple of the wrong arity or incompatible types raises an error; here that means a two-element tuple matching `count` and `last_seen`.
• D. Timeout callbacks only fire when `timeoutConf` enables them; with `NoTimeout()` selected, Spark never invokes the function purely because time has elapsed, so this property stays `False` on every call.

Correct answer: A — Insert `result = result.coalesce(8)` before the write, since it merges existing partitions on the same executors and avoids a full shuffle across the cluster.

• A. `coalesce()` reduces the number of partitions by combining existing ones on the same executors, so it can shrink partition count without a full shuffle when the target is fewer partitions than the current count. This makes it the cheapest way to hit exactly 8 output files here.
• B. `repartition()` always performs a full shuffle to redistribute rows across the new partition count, regardless of direction, so it is more expensive than needed when only reducing the number of partitions.
• C. `spark.sql.shuffle.partitions` only controls the number of partitions produced by shuffle operations such as joins and aggregations; it has no effect on the partition count of a plain filter-then-write DataFrame operation.
• D. Hash partitioning on a single low-cardinality column like `region` does not guarantee balanced output; rows with the same region hash to the same partition, so file sizes can still be very uneven, and this still forces a full shuffle.
• E. Caching a DataFrame stores its existing partitions in memory or on disk; it does not change the partition count at all, so the write would still emit the original number of files.

Correct answer: A — `spark.conf.set("spark.sql.shuffle.partitions", "auto")` starts each shuffle with a large partition count and lets AQE coalesce it down toward the advisory partition size.

• A. This is correct because setting the shuffle partition count to `"auto"` on Databricks starts with a large initial partition count and delegates the final sizing to AQE's coalescing logic based on actual shuffled data volume.
• B. This is incorrect because the coalescing enabled flag is a plain boolean; it does not accept an `"auto"` value that conditionally toggles behavior by input size.
• C. This is incorrect because setting the partition count to zero is not a supported way to derive partitions from executor core count; it does not enable automatic AQE-driven sizing.
• D. This is incorrect because disabling this Databricks-specific auto-optimization flag does the opposite of what is requested, forcing a fixed manual partition count instead of automatic sizing.
• E. This is incorrect because this setting controls input file read partitioning, not the number of partitions used for shuffles triggered later by joins or aggregations in the query.

Correct answers: A, B — Remove the explicit `broadcast(dim)` hint so the join falls back to a standard sort-merge join, since forcing a 12 GB table to be broadcast to every executor is what is overwhelming executor memory regardless of the threshold setting.; Reduce `dim_store` to only the columns and rows the join actually needs before the join executes, and confirm the filtered DataFrame is genuinely small so the existing `broadcast(dim)` hint can safely execute without overwhelming any executor.

• A. An explicit `broadcast()` hint forces Spark to ship the entire hinted DataFrame to every executor regardless of the `autoBroadcastJoinThreshold` setting, so removing the hint lets the optimizer fall back to a standard sort-merge join that processes `dim_store` in shuffled partitions instead of copying all 12 GB into every executor's memory.
• B. The underlying problem is that `dim_store` grew past the point where broadcasting it is safe, so trimming it down to only the join-relevant columns and rows — and verifying the result is actually small — restores the assumption the original code was written under and lets the broadcast hint execute without exhausting executor memory.
• C. `spark.sql.shuffle.partitions` controls partitioning for shuffle-based joins such as sort-merge joins, but an explicit `broadcast()` hint bypasses the shuffle path entirely and copies the whole DataFrame to every executor, so raising this setting has no effect on the broadcast operation that is actually failing.
• D. Setting a storage level on `dim_store` controls how that DataFrame's data is cached for reuse, but it has no relationship to the broadcast join threshold or to how the explicit `broadcast()` hint decides to ship data to executors, so this change would not prevent the same broadcast operation from failing.
• E. A cross join followed by a filter still requires materializing the full pairwise combination of both tables' partitions on the executors that process them, which is typically far more expensive than either a sort-merge join or a safely sized broadcast, and does not avoid memory pressure — it usually makes it worse.

Correct answer: A — Spark uses the default micro-batch trigger, immediately starting the next micro-batch as soon as the previous one finishes processing

• A. This is correct: when no trigger is specified, Spark uses the default micro-batch trigger, which starts a new batch as soon as the prior one finishes, giving the lowest latency the micro-batch engine can offer without explicit tuning.
• B. This is incorrect because continuous processing is an experimental, opt-in execution mode requiring `.trigger(continuous=...)`; omitting a trigger does not switch the engine into that mode.
• C. This is incorrect because the absence of a trigger clause does not pause the query; Spark falls back to its default trigger behavior rather than waiting for configuration.
• D. This is incorrect because processing a single batch and stopping is the behavior of a one-time trigger such as `Trigger.Once()`, not the default trigger that is used when no trigger is specified.

Correct answer: A — Micro-batch mode polls the source and batches accumulated data, while continuous processing runs long tasks that handle records as they arrive for lower latency

• A. This is correct: micro-batch execution repeatedly polls the source and processes whatever has accumulated in discrete batches, while continuous processing keeps long-running tasks reading and processing records individually to achieve much lower latency.
• B. This is incorrect because cluster autoscaling behavior is independent of which streaming execution mode is chosen; neither mode inherently fixes or dynamically resizes the cluster per record.
• C. This is incorrect because micro-batch mode supports a wide range of sinks including Delta tables and JDBC, not only files; sink support is not the distinguishing factor between the two modes.
• D. This is incorrect because micro-batch mode is the execution model that provides Structured Streaming's well-supported exactly-once guarantees, while continuous processing offers weaker, at-least-once guarantees for its supported operations.

Correct answer: A — `session_window()` builds one window per group that keeps extending while new events arrive within the gap duration, and finalizes once that inactivity gap has elapsed.

• A. This is correct because `session_window` is purpose-built for dynamic-length windows keyed on a gap duration: a session stays open while new events keep arriving within that gap and is finalized once the gap is exceeded, matching the requirement exactly.
• B. This is incorrect because a tumbling window has a fixed boundary aligned to the clock regardless of activity, so it would split or merge events based on wall-clock time rather than actual inactivity gaps.
• C. This is incorrect because tracking only the maximum event time per user produces a single running aggregate value, not distinct session windows that open and close based on gaps between events.
• D. This is incorrect because sliding windows still have fixed durations and slide intervals tied to the clock, so they do not dynamically expand or contract based on the actual gap between a user's events.
• E. This is incorrect because `dropDuplicatesWithinWatermark` removes duplicate rows within a watermark bound; it does not group events into variable-length sessions based on inactivity gaps.

Correct answers: B, C — `update`; `complete`

• A. Stream-to-stream joins are only supported with the `append` output mode, so this mode is valid and does not raise an exception.
• B. `update` mode is not yet supported for stream-to-stream joins; passing it to `.outputMode()` for a join query raises an `AnalysisException`.
• C. `complete` mode is not supported for stream-to-stream joins either, since the join does not produce a bounded result table that can be rewritten in full each trigger.
• D. `once` is a trigger setting, not an output mode string, so it is not a candidate answer to this question about `.outputMode()` values.
• E. `default` is not a recognized output mode identifier in Structured Streaming; the valid mode names are `append`, `update`, and `complete`.

Correct answer: A — The state for that key was already evicted once the watermark passed 10:00:00 by more than the 15-minute delay, so the engine can no longer recognize the duplicate and the record passes through as new.

• A. This is correct: once the watermark advances past a key's event time by more than the declared delay, the engine treats that key's state as safe to remove, so a genuine duplicate arriving after that point is no longer matched and is emitted as a new row. This is the expected tradeoff of bounding state with a watermark.
• B. There is no secondary unbounded index kept outside the state store; the watermark's entire purpose is to let the engine physically discard state, and once discarded there is no other record of the key to check against.
• C. Structured Streaming does not throw an exception for late data relative to the watermark; late records that fall outside the bound are simply treated as new input rather than causing the query to fail.
• D. The watermark delay is a fixed configuration set when the query is defined; the engine does not dynamically extend it backward in response to individual late arrivals, since that would defeat the purpose of bounding state at all.
• E. There is no buffering-until-checkpoint mechanism that reconciles late records against deleted keys; the state for expired keys is gone, and buffering indefinitely would reintroduce the unbounded growth the watermark is meant to prevent.

Correct answer: A — The threshold should exceed the maximum expected gap between an event and its latest duplicate, so genuine duplicates are not mistaken for new records once the watermark passes.

• A. This is correct: the watermark delay directly determines how long state is retained for a given key, so it must cover the realistic lateness of duplicate arrivals or the engine will expire state too early and fail to catch genuine duplicates.
• B. The watermark threshold directly controls state retention and correctness of duplicate detection, not merely how often UI statistics are refreshed, which is an unrelated concern from query configuration.
• C. The watermark delay is an independently configured duration passed to `withWatermark` and is not derived from or required to match the micro-batch trigger interval, which controls how often batches are processed.
• D. The watermark threshold is expressed as a time duration relative to event time, such as minutes or hours, not as a count of expected retries per key.
• E. The watermark threshold governs state retention for any stateful operator that uses it, including deduplication, not only aggregation operators; deduplication state expiry depends directly on this same threshold.

Correct answers: A, B — The thin client library only encodes DataFrame operations into unresolved logical plans; it does not depend on the internal execution engine version running on the server.; Spark Connect's gRPC protocol maintains compatibility across versions, so an older client library can keep issuing requests to a newer server without a matching client-side upgrade.

• A. Correct — because the client only builds and transmits an unresolved logical plan, everything version-specific about execution stays on the server side, so a server upgrade does not change what the client needs to send.
• B. Correct — Spark Connect is designed so the gRPC-based protocol between client and server stays compatible across versions, letting an existing client keep working against a newly upgraded server without needing its own rebuild.
• C. Incorrect — there is no shared JVM classpath exchanged between client and server; Spark Connect deliberately avoids that kind of tight coupling, which is why client and server can be upgraded independently.
• D. Incorrect — clients submit DataFrame operations as plans, not compiled Java bytecode, so there is no bytecode-to-intermediate-representation conversion happening as part of a server upgrade.
• E. Incorrect — Spark Connect does not pin clients to a fixed server version; the whole point of protocol compatibility is that a client can keep talking to a server that has since been upgraded.

Correct answer: A — Databricks Connect acts as a Spark Connect client: DataFrame operations build a logical plan locally and send it over gRPC to the remote cluster for execution.

• A. Databricks Connect is built as a Spark Connect client, so DataFrame code written locally builds a logical plan that is sent over gRPC to the remote Databricks cluster, which executes it with access to Unity Catalog. This matches the intended workflow.
• B. Databricks Connect does not install or copy the Databricks Runtime onto the local machine; the whole point of the thin client design is that execution happens remotely, not that the runtime is duplicated locally.
• C. One of the main benefits of Databricks Connect is that it lets a local IDE submit Spark Connect requests directly to a remote cluster without ever converting the script into a notebook first.
• D. Running both the driver and executors locally would make Databricks Connect indistinguishable from local mode, but its purpose is remote execution against a real Databricks cluster and its Unity Catalog access, not local computation.
• E. Databricks Connect submits work through the Spark execution engine via Spark Connect, not by rewriting DataFrame operations into SQL warehouse queries; the underlying engine and plan structure remain Spark's own.

Correct answer: A — `pyspark.pandas` is a pandas-compatible layer built on top of Spark DataFrames, so operations written with its API are translated into the same distributed Spark execution plan that native PySpark DataFrames use.

• A. This is correct: `pyspark.pandas` provides pandas-compatible syntax that is executed through the same underlying Spark DataFrame execution engine, so it is a layer on top of, not a replacement for, native PySpark DataFrames.
• B. This is incorrect because `pyspark.pandas` runs within the Spark cluster and relies on the same distributed engine as native DataFrames; it is not a separate single-node execution engine outside the cluster.
• C. This is incorrect because the native `pyspark.sql.DataFrame` API remains fully supported and is not deprecated; `pyspark.pandas` is an additional, complementary interface rather than a replacement.
• D. This is incorrect because the pandas API on Spark is designed to run on distributed multi-node clusters just like native PySpark, and there is no local-mode-only restriction or automatic fallback behavior.
• E. This is incorrect because `pyspark.pandas` provides a full set of DataFrame transformation, aggregation, and I/O methods mirroring pandas; plotting is only one small part of its functionality, not its entire purpose.

Correct answer: A — ```python def clean_batch(iterator): for pdf in iterator: pdf = pdf[pdf["amount"] >= 0].copy() pdf["processed_at"] = pd.Timestamp.now() yield pdf df.mapInPandas(clean_batch, schema=new_schema) ```, applied directly on the ungrouped DataFrame.

• A. mapInPandas takes a function that accepts and returns an iterator of pandas DataFrames representing arbitrary batches, and requires an explicit schema argument describing the output; calling it directly on the ungrouped DataFrame matches exactly how this batch-level filtering and column addition is meant to be expressed.
• B. mapInPandas processes arbitrary batches of the whole DataFrame and is not a GroupedData method, so chaining it after groupby is not a supported call pattern for this operation.
• C. mapInPandas expects a plain Python function with an iterator-to-iterator signature, not a function decorated with pandas_udf; wrapping the function this way and calling mapInPandas without a schema argument does not match its required calling convention.
• D. mapInPandas requires the function to accept and yield an iterator of pandas DataFrames, not a single DataFrame passed and returned directly, so this signature does not match what mapInPandas expects to call.
• E. mapInPandas requires the schema to be passed explicitly as an argument because Spark cannot infer the output schema from Python code before execution; omitting it raises an error rather than being inferred automatically.