To enable fault tolerance and ensure that Spark can resume from the last committed offset after failure, you must configure a checkpoint location using the correct option key: "checkpointLocation".
From the official Spark Structured Streaming guide:
"To make a streaming query fault-tolerant and recoverable, a checkpoint directory must be specified using .option("checkpointLocation", "/path/to/dir")." Explanation of options:
Option A uses an invalid option name: "checkpoint" (should be "checkpointLocation") Option B is correct: it sets checkpointLocation properly Option C lacks checkpointing and won't resume after failure Option D also lacks checkpointing configuration

The Spark UI is the standard and most effective way to inspect executor logs, task time, input size, and shuffles.
From the Databricks documentation:
"You can monitor job execution via the Spark Web UI. It includes detailed logs and metrics, including task-level execution time, shuffle reads/writes, and executor memory usage." (Source: Databricks Spark Monitoring Guide) Option A is incorrect: logs are not guaranteed to be in /tmp, especially in cloud environments.
B . -verbose helps during job submission but doesn't give detailed executor logs.
D . spark-sql is a CLI tool for running queries, not for inspecting logs.
Hence, the correct method is using the Spark UI → Stages tab → Executor logs.

According to Spark's watermarking rules:
"Records that are older than the watermark (event time < current watermark) are considered too late and are dropped." So, if a record's event_time is earlier than (max event_time seen so far - 10 minutes), it is discarded.

To ensure that data written out to disk is sorted, it is important to consider how Spark writes data when saving to Parquet tables. The methods .sort() or .orderBy() apply a global sort but do not guarantee that the sorting will persist in the final output files unless certain conditions are met (e.g. a single partition via .coalesce(1) - which is not scalable).
Instead, the proper method in distributed Spark processing to ensure rows are sorted within their respective partitions when written out is:
.sortWithinPartitions("column_name")
According to Apache Spark documentation:
"sortWithinPartitions() ensures each partition is sorted by the specified columns. This is useful for downstream systems that require sorted files." This method works efficiently in distributed settings, avoids the performance bottleneck of global sorting (as in .orderBy() or .sort()), and guarantees each output partition has sorted records - which meets the requirement of consistently sorted data.
Thus:
Option A and B do not guarantee the persisted file contents are sorted.
Option C introduces a bottleneck via .coalesce(1) (single partition).
Option D correctly applies sorting within partitions and is scalable.

The correct answer is D because it uses proper time-based window aggregation along with watermarking, which is the required pattern in Spark Structured Streaming for time-based aggregations over event-time data.
From the Spark 3.5 documentation on structured streaming:
"You can define sliding windows on event-time columns, and use groupBy along with window() to compute aggregates over those windows. To deal with late data, you use withWatermark() to specify how late data is allowed to arrive." (Source: Structured Streaming Programming Guide) In option D, the use of:
python
CopyEdit
.groupBy("sensor_id", window("timestamp", "5 minutes"))
.agg(avg("temperature").alias("avg_temp"))
ensures that for each sensor_id, the average temperature is calculated over 5-minute event-time windows. To complete the logic, it is assumed that withWatermark("timestamp", "5 minutes") is used earlier in the pipeline to handle late events.
Explanation of why other options are incorrect:
Option A uses Window.partitionBy which applies to static DataFrames or batch queries and is not suitable for streaming aggregations.
Option B does not apply a time window, thus does not compute the rolling average over 5 minutes.
Option C incorrectly applies withWatermark() after an aggregation and does not include any time window, thus missing the time-based grouping required.
Therefore, Option D is the only one that meets all requirements for computing a time-windowed streaming aggregation.