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'sevent_timeis earlier than (max event_time seen so far - 10 minutes), it is discarded.
Reference:Structured Streaming - Handling Late Data

Comprehensive and Detailed Explanation:
Exactly-once guarantees in Spark Structured Streaming require micro-batch mode (default), not continuous mode.
Continuous mode (.trigger(continuous=...)) only supports at-least-once semantics and lacks full fault- tolerance.
trigger(availableNow=True)is a batch-style trigger, not suited for low-latency streaming.
So:
Option A uses micro-batching with a tight trigger interval # minimal latency + exactly-once guarantee.
Final Answer: A

Comprehensive and Detailed Explanation From Exact Extract:
In Spark, the default join type is an inner join, which returns only the rows with matching keys in both DataFrames. To retain all records from the left DataFrame (purch_df) and include matching records from the right DataFrame (cust_df), a left outer join should be used.
By specifying the join type as'left', the modified code ensures that all records frompurch_dfare preserved, and matching records fromcust_dfare included. Records inpurch_dfwithout a corresponding match incust_dfwill havenullvalues for the columns fromcust_df.
This approach is consistent with standard SQL join operations and is supported in PySpark's DataFrame API.

Comprehensive and Detailed Explanation From Exact Extract:
The number of tasks is controlled by the number of partitions. By default,spark.sql.shuffle.partitionsis 200. If stages are showing very few tasks (less than total cores), you may not be leveraging full parallelism.
From the Spark tuning guide:
"To improve performance, especially for large clusters, increasespark.sql.shuffle.partitionsto create more tasks and parallelism." Thus:
A is correct: increasing shuffle partitions increases parallelism
B is wrong: it further reduces parallelism
C is invalid: increasing dataset size doesn't guarantee more partitions D is irrelevant to task count per stage Final Answer: A

Spark uses lazy evaluation. Transformations like filter, select, and groupBy only define the DAG (Directed Acyclic Graph). No execution occurs until an action is triggered.
The first action in the code is:reduced_df.count()
So Spark starts processing data at this line.
Reference:Apache Spark Programming Guide - Lazy Evaluation