1. Concept Overview The standard and most robust design pattern for deduplicating records by timestamp in Apache Spark is applying the **`row_number()` window function**. By partitioning the dataset by the business entity key (`customer_id`), ordering the records within each partition by the timestamp descending (`update_date.desc()`), and filtering for `row_number == 1`, the pipeline deterministically isolates the single most recent record per customer.

2. Architecture & Design Principle 1. **Window-Based Deduplication vs. GroupBy**: - Using `groupBy("customer_id").agg(max("update_date"))` only returns the key and the latest timestamp; it drops all other non-aggregate payload columns (like `city`) unless complex struct aggregations are used. - `row_number()` over `Window.partitionBy("customer_id").orderBy(col("update_date").desc())` preserves the entire schema, attaching an integer rank to each row. 2. **Deterministic Partitioning**: Spark shuffles rows with identical `customer_id` to the same partition, sorts them in-memory by `update_date` descending, and assigns `1` to the newest record.

3. Step-by-Step Code / Logic Execution 1. Construct the window specification:

   from pyspark.sql.functions import col, row_number
   from pyspark.sql.window import Window

   window_spec = Window.partitionBy("customer_id").orderBy(col("update_date").desc())

2. Apply the window function and filter:

   latest_customers = (
       df.withColumn("rn", row_number().over(window_spec))
         .filter(col("rn") == 1)
         .drop("rn")
   )

3. Execution trace: - For customer C101 with updates on '2026-01-01' and '2026-01-15': - '2026-01-15' receives rn = 1 (retained). - '2026-01-01' receives rn = 2 (filtered out).

4. Why the Correct Answer Is Optimal Option C is correct because: - `row_number()` guarantees a unique integer for each record, preventing duplicate rows even when two updates share identical timestamps. - Filtering on `rn == 1` guarantees exactly one record per customer entity.

5. Why Alternative Options Fail - Option A (min()): Incorrect; `min()` computes the minimum scalar value, not a row ranking index. - Option B (avg()): Incorrect; `avg()` computes an arithmetic mean. - Option D (count()): Incorrect; `count()` calculates row totals, not relative sequential ordering.

6. Edge Cases & Production Pitfalls - **Tied Timestamps**: If two updates for the same customer share the exact same millisecond timestamp, `row_number()` arbitrarily picks one unless a secondary sort key is added: `orderBy(col("update_date").desc(), col("ingestion_id").desc())`. - **Global Memory Overhead**: If a single customer has 10 million updates, that entire partition must fit into executor memory during window evaluation.

7. Performance & Optimization Considerations - In Spark 3.0+, Catalyst optimizes window row-number filters by performing **window rank pushdown**, discarding rows with rank > 1 early in the physical plan.

8. Reliability & Operational Best Practices - Always drop the temporary ranking column (`.drop("rn")`) before persisting to production tables to maintain schema cleanliness.

9. Real-World Architectural Scenario A financial CRM consolidates customer address updates from mobile apps, call centers, and branch visits. Using `row_number()` partitioned by `customer_id` ordered by `update_timestamp.desc()` ensures that marketing communications are dispatched to the customer's verified latest address.

10. Official Documentation & Learning Resources - Window Deduplication: [PySpark DataFrame Window Functions Tutorial](/tutorials/pyspark/df-window-functions)

1. Concept Overview The "Top N per Category" problem is one of the most common distributed analytical patterns in data engineering. In PySpark, the optimal, scalable approach is defining a **Window partitioned by the category** (`branch_id`), **ordered by the ranking metric descending** (`balance.desc()`), assigning a sequential row number with **`row_number()`**, and filtering where `row_number <= 5`.

2. Architecture & Design Principle 1. **Distributed Partitioning**: - `Window.partitionBy("branch_id")` distributes accounts across the cluster by branch. - Each branch's accounts are sorted independently in parallel by executor cores. 2. **Cluster Scalability vs. Driver Bottlenecks**: - A naive approach using `collect()` pulls all accounts to the single Spark Driver program, instantly causing Out-Of-Memory (OOM) crashes on large datasets. - The Window approach executes entirely within Spark's distributed memory fabric across all worker nodes.

3. Step-by-Step Code / Logic Execution 1. Define the branch window specification:

   from pyspark.sql.functions import col, row_number
   from pyspark.sql.window import Window

   branch_window = Window.partitionBy("branch_id").orderBy(col("balance").desc())

2. Compute row rankings and filter top 5:

   top5_accounts = (
       df.withColumn("rank", row_number().over(branch_window))
         .filter(col("rank") <= 5)
         .drop("rank")
   )

3. Execution trace: - For Branch B01: 50,000 accounts are sorted. Accounts 1 through 5 are emitted; accounts 6 through 50,000 are dropped. - For Branch B02: Evaluated simultaneously on another worker node.

4. Why the Correct Answer Is Optimal Option A is correct because: - It solves the Top-N problem in distributed parallel fashion without centralizing data on the driver. - It guarantees exactly 5 records per branch (or fewer if a branch has less than 5 accounts).

5. Why Alternative Options Fail - Option B (collect()): Incorrect; calling `collect()` on millions of customer balances crashes the driver node with an Out-Of-Memory error. - Option C (distinct()): Incorrect; `distinct()` only removes duplicate rows; it cannot perform ranking or numeric thresholding. - Option D (union()): Incorrect; `union()` concatenates DataFrames together; it does not rank records within groups.

6. Edge Cases & Production Pitfalls - **Ties in Balance**: If accounts rank 5 and 6 have identical balances, `row_number()` picks one arbitrarily. If business policy mandates that ties must all be included, use `dense_rank() <= 5`.

7. Performance & Optimization Considerations - When there are thousands of branches, `partitionBy("branch_id")` creates an even distribution of data across shuffle partitions.

8. Reliability & Operational Best Practices - Add an explicit tie-breaking column (e.g., `orderBy(col("balance").desc(), col("account_id").asc())`) to ensure deterministic top-5 output across pipeline re-runs.

9. Real-World Architectural Scenario A national wealth management firm generates monthly VIP account reports for 450 branch managers. Executing `row_number().over(Window.partitionBy("branch_id").orderBy(col("portfolio_value").desc())).filter(col("rank") <= 5)` processes 30 million customer records in 90 seconds.

10. Official Documentation & Learning Resources - Window Functions: [PySpark Window Functions & Top-N Analysis](/tutorials/pyspark/df-window-functions)

1. Concept Overview The **`lag()` window function** accesses data from a preceding row at a specified physical offset within an ordered window partition. To compare a customer's current monthly bandwidth usage against their previous month's usage, `lag("usage", 1)` looks backward by exactly one month in the time-sorted partition, making it trivial to calculate month-over-month usage deltas.

2. Architecture & Design Principle 1. **Partitioned Chronological Ordering**: - Data is partitioned by subscriber (`subscriber_id`) and sorted ascending by billing month (`billing_month.asc()`). - For each row, Spark accesses the cached preceding row in the partition memory buffer. 2. **Avoiding Expensive Self-Joins**: - In traditional SQL without window functions, computing previous month usage requires an expensive self-join: `df.join(df, df.month == df_prev.month + 1)`. - `lag()` performs the lookback in a single shuffle pass with zero self-join overhead.

3. Step-by-Step Code / Logic Execution 1. Define the window partitioned by customer:

   from pyspark.sql.functions import col, lag
   from pyspark.sql.window import Window

   user_window = Window.partitionBy("subscriber_id").orderBy("billing_month")

2. Retrieve previous month usage and calculate difference:

   usage_trend_df = (
       df.withColumn("prev_month_usage", lag("usage_gb", 1).over(user_window))
         .withColumn("usage_diff", col("usage_gb") - col("prev_month_usage"))
   )

3. Execution trace for subscriber S1: - Jan 2026: Usage = 50 GB -> Prev = NULL -> Diff = NULL. - Feb 2026: Usage = 80 GB -> Prev = 50 GB -> Diff = +30 GB. - Mar 2026: Usage = 70 GB -> Prev = 80 GB -> Diff = -10 GB.

4. Why the Correct Answer Is Optimal Option A is correct because: - `lag()` is specifically designed for retrospective offset lookups. - It operates with optimal linear memory efficiency within each partition.

5. Why Alternative Options Fail - Option B (lead()): Incorrect; `lead()` looks forward to subsequent months (future usage), not previous months. - Option C (sum()): Incorrect; `sum()` aggregates usage, producing totals rather than prior-period values. - Option D (count()): Incorrect; `count()` returns the number of months.

6. Edge Cases & Production Pitfalls - **Missing Months**: `lag("usage", 1)` looks back by one *physical row*, not one *calendar month*. If a customer had no usage in February, the row preceding March will be January. To enforce strict calendar spacing, verify date differences or pre-fill missing months.

7. Performance & Optimization Considerations - Always include `partitionBy("subscriber_id")` so that subscribers are distributed across all executor nodes.

8. Reliability & Operational Best Practices - Provide a default value: `lag("usage_gb", 1, 0.0)` ensures the first recorded month displays 0 instead of `NULL`.

9. Real-World Architectural Scenario A telecom provider alerts customers whose data consumption surges by more than 50% compared to the prior month. Calculating `col("current_usage") - lag("current_usage", 1).over(w)` powers real-time overage alert pipelines for 40 million subscribers.

10. Official Documentation & Learning Resources - Time-Series Analysis: [PySpark Window Functions & Lag Operations](/tutorials/pyspark/df-window-functions)

1. Concept Overview To identify duplicate composite keys in a dataset, the standard data engineering solution is grouping by the business key columns using **`groupBy("customer_id", "order_id")`**, calculating the occurrence count with **`.count()`**, and filtering for groups where the frequency exceeds one (`.filter(col("count") > 1)`).

2. Architecture & Design Principle 1. **Map-Reduce Frequency Aggregation**: - Spark hashes the composite key `(customer_id, order_id)` and routes matching records to the same shuffle partition. - Each partition aggregates counts locally (`partial_count`), followed by a final reduction across partitions. 2. **Data Quality Audit Pattern**: This pattern isolates the offending duplicate keys along with their exact repetition frequency, enabling engineers to quantify the severity of upstream ingestion bugs.

3. Step-by-Step Code / Logic Execution 1. Group by composite key and aggregate count:

   from pyspark.sql.functions import col

   duplicate_keys_df = (
       df.groupBy("customer_id", "order_id")
         .count()
         .filter(col("count") > 1)
   )

2. Execution trace: - Key ('C1', 'O100') appears 1 time -> filtered out. - Key ('C2', 'O200') appears 3 times -> emitted with count = 3. - Key ('C3', 'O300') appears 1 time -> filtered out. 3. The resulting DataFrame contains only the duplicate keys and their duplicate frequency.

4. Why the Correct Answer Is Optimal Option D is correct because: - `groupBy().count()` provides both identification of duplicate keys and their exact frequency. - It executes entirely as a distributed aggregation across all cluster nodes.

5. Why Alternative Options Fail - Option A (collect()): Incorrect; pulling multi-terabyte data to the driver program crashes the JVM. - Option B (cache()): Incorrect; caching stores data in memory; it performs no analysis or duplicate detection. - Option C (coalesce()): Incorrect; `coalesce()` reduces the number of partitions without shuffling; it does not identify duplicates.

6. Edge Cases & Production Pitfalls - **Null Values in Composite Keys**: If `order_id` contains NULL values, `groupBy()` treats `(C1, NULL)` as a valid group. If multiple rows have `(C1, NULL)`, they will be flagged as duplicates. Filter out or explicitly handle NULLs if unassigned orders are expected.

7. Performance & Optimization Considerations - If you only want to know whether duplicates exist rather than inspecting them, `df.count() != df.dropDuplicates(["customer_id", "order_id"]).count()` provides a quick boolean check.

8. Reliability & Operational Best Practices - Integrate duplicate key checks into automated CI/CD data quality assertions (e.g., Great Expectations or Soda Core) before publishing to Gold analytics layers.

9. Real-World Architectural Scenario An e-commerce order ingestion pipeline experiences message redelivery from Kafka. Running the `groupBy("customer_id", "order_id").count().filter("count > 1")` validation check catches order duplication before credit cards are double-charged.

10. Official Documentation & Learning Resources - Aggregation Patterns: [PySpark Aggregations & Data Validation Guide](/tutorials/pyspark/pyspark-aggregation)

1. Concept Overview When business requirements specify finding the "N-th highest distinct value" (such as the 3rd highest salary) where ties must not skip ranking numbers, **`dense_rank()`** is the required function. Because `dense_rank()` produces consecutive, gapless rank numbers (1, 1, 2, 2, 3), filtering on `dense_rank == 3` reliably extracts the 3rd distinct salary tier regardless of how many employees share the 1st or 2nd highest salaries.

2. Architecture & Design Principle 1. **Why `rank()` Fails for Value Tiers**: - If 3 executives tie for the highest salary ($500k), `rank()` assigns them `1, 1, 1` and skips ranks 2 and 3; the next highest salary ($400k) receives rank `4`. - Filtering for `rank == 3` would return **zero rows**, completely missing the 3rd highest salary! 2. **Why `dense_rank()` Succeeds**: - The top 3 executives receive `dense_rank = 1`. - The next tier ($400k) receives `dense_rank = 2`. - The third distinct tier ($350k) receives `dense_rank = 3`.

3. Step-by-Step Code / Logic Execution 1. Define the descending order window:

   from pyspark.sql.functions import col, dense_rank
   from pyspark.sql.window import Window

   salary_window = Window.orderBy(col("salary").desc())

2. Assign dense rank and filter for the 3rd tier:

   third_highest_salary_df = (
       df.withColumn("drank", dense_rank().over(salary_window))
         .filter(col("drank") == 3)
   )

3. Execution trace on salaries [5000, 5000, 4000, 3000]: - 5000 -> dense_rank = 1 - 5000 -> dense_rank = 1 - 4000 -> dense_rank = 2 - 3000 -> dense_rank = 3 (Matched!)

4. Why the Correct Answer Is Optimal Option B is correct because: - `dense_rank()` numbers distinct salary levels without skipping integers. - It guarantees that the 3rd distinct salary level is identified accurately.

5. Why Alternative Options Fail - Option A (lead()): Incorrect; `lead()` looks forward to subsequent rows; it does not compute ranks. - Option C (sum()): Incorrect; `sum()` computes numeric addition. - Option D (lag()): Incorrect; `lag()` looks backward to preceding rows.

6. Edge Cases & Production Pitfalls - **Fewer Than N Tiers**: If the dataset contains only 2 distinct salary values, `filter(col("drank") == 3)` returns an empty DataFrame. Always verify row count before consuming results.

7. Performance & Optimization Considerations - If you only need the scalar 3rd highest salary rather than the employee rows, `df.select("salary").distinct().orderBy(col("salary").desc()).limit(3).tail(1)` avoids window evaluation overhead.

8. Reliability & Operational Best Practices - Distinguish between "3rd highest employee" (`row_number == 3`) and "3rd highest salary tier" (`dense_rank == 3`).

9. Real-World Architectural Scenario An HR analytics platform generates annual compensation benchmark reports for 250,000 global employees. Using `dense_rank()` partitioned by job role and grade identifies the exact salary tiers for standardized compensation bands.

10. Official Documentation & Learning Resources - Window Functions: [PySpark Window Functions & Dense Rank Tutorial](/tutorials/pyspark/df-window-functions)

1. Concept Overview Computing a cumulative running total across a sequence of chronological events requires calculating a progressive sum over an ordered partition. In PySpark, this is achieved using **`Window + sum()`** with `partitionBy("customer_id")`, `orderBy("transaction_date")`, and an explicit frame definition (`rowsBetween(Window.unboundedPreceding, Window.currentRow)`).

2. Architecture & Design Principle 1. **Partitioned Progressive Accumulation**: - Standard `groupBy()` collapses all rows for a customer into a single scalar sum, discarding the individual transaction dates. - `Window + sum()` preserves every original row and appends a new column containing the running sum up to that transaction date. 2. **Distributed Execution**: Each customer's transactions are assigned to an executor partition and sorted by date, allowing millions of customer running totals to be computed concurrently in parallel.

3. Step-by-Step Code / Logic Execution 1. Define the customer window:

   from pyspark.sql.functions import col, sum
   from pyspark.sql.window import Window

   running_total_window = (
       Window.partitionBy("customer_id")
             .orderBy("transaction_date")
             .rowsBetween(Window.unboundedPreceding, Window.currentRow)
   )

2. Compute the running total:

   result_df = df.withColumn("running_total", sum("amount").over(running_total_window))

3. Execution trace for customer C1: - Day 1: Amount = 100 -> Running Total = 100. - Day 2: Amount = 250 -> Running Total = 350. - Day 3: Amount = 50 -> Running Total = 400.

4. Why the Correct Answer Is Optimal Option C is correct because: - Windowing preserves row-level granularity while computing progressive cumulative sums. - It provides linear computational complexity per partition without multi-pass scanning.

5. Why Alternative Options Fail - Option A (groupBy()): Incorrect; `groupBy()` collapses multiple rows into a single aggregated row, losing individual transaction dates. - Option B (distinct()): Incorrect; `distinct()` removes duplicate rows. - Option D (collect()): Incorrect; `collect()` pulls all raw records to the driver memory, causing driver OOM crashes.

6. Edge Cases & Production Pitfalls - **Default Range Framing with Duplicate Timestamps**: If multiple transactions occur on the same day and default framing is used, Spark's default `RANGE` frame sums all same-day transactions simultaneously into both rows. Using explicit `rowsBetween(Window.unboundedPreceding, Window.currentRow)` ensures progressive step accumulation.

7. Performance & Optimization Considerations - Always include `partitionBy("customer_id")` to prevent all cluster data from being routed to a single executor partition.

8. Reliability & Operational Best Practices - Specify `rowsBetween` explicitly in all production cumulative window definitions to ensure deterministic behavior.

9. Real-World Architectural Scenario A credit card banking engine displays transaction history in mobile banking apps. Each purchase or payment appends the updated account balance using `sum("amount").over(Window.partitionBy("account_id").orderBy("txn_time").rowsBetween(unboundedPreceding, currentRow))`.

10. Official Documentation & Learning Resources - Window Aggregations: [PySpark Window Functions & Cumulative Sums Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview In distributed data pipelines, out-of-order and late-arriving records are common. When reconciling multiple versions of an entity record to retain the most recent state, the industry standard technique in PySpark is applying **`row_number()` ordered by timestamp descending** within a window partitioned by the primary entity key (`encounter_id`), followed by filtering for `row_number == 1`.

2. Architecture & Design Principle 1. **Idempotent State Reconstruction**: - The primary key is `encounter_id`. - Ordering by `record_timestamp.desc()` ensures that the newest diagnosis version always occupies index position 1. - Filtering on `row_number == 1` prunes all stale, superseded versions. 2. **Full-Row Preservation**: Unlike aggregate functions, this approach preserves all clinical diagnosis codes, physician notes, and medication lists from the latest record without complex join reconstruction.

3. Step-by-Step Code / Logic Execution 1. Construct the window:

   from pyspark.sql.functions import col, row_number
   from pyspark.sql.window import Window

   encounter_window = (
       Window.partitionBy("encounter_id")
             .orderBy(col("record_timestamp").desc())
   )

2. Filter for latest state:

   latest_encounters = (
       df.withColumn("rn", row_number().over(encounter_window))
         .filter(col("rn") == 1)
         .drop("rn")
   )

3. Execution trace: - Encounter E901 arrived at 10:00 AM (Diagnosis: Flu). - Late update for E901 arrived at 02:00 PM (Diagnosis: Pneumonia). - The 02:00 PM record receives rn = 1 and is preserved; the 10:00 AM record receives rn = 2 and is dropped.

4. Why the Correct Answer Is Optimal Option A is correct because: - Ordering by descending timestamp guarantees that the newest record receives rank 1. - `row_number()` guarantees that exactly one record is retained even if multiple updates share the same timestamp.

5. Why Alternative Options Fail - Option B (count()): Incorrect; `count()` returns frequencies, not records. - Option C (union()): Incorrect; `union()` concatenates data and would preserve duplicates rather than removing them. - Option D (min()): Incorrect; `min()` on timestamp would keep the oldest record, the exact opposite of the requirement.

6. Edge Cases & Production Pitfalls - **Tie-Breaking**: If two clinic updates have identical timestamps, add a secondary tie-breaker: `orderBy(col("record_timestamp").desc(), col("source_sequence").desc())`.

7. Performance & Optimization Considerations - Spark's Catalyst engine applies filter pushdown on window row numbers, avoiding full buffer materialization for superseded rows.

8. Reliability & Operational Best Practices - Log the number of discarded superseded records to track clinic sync latency and late-arrival rates.

9. Real-World Architectural Scenario A hospital electronic health record (EHR) pipeline processes 5 million daily patient updates. Re-evaluating encounters with `row_number()` over `record_timestamp.desc()` ensures ICU doctors see the latest lab and vitals data.

10. Official Documentation & Learning Resources - Data Cleansing: [PySpark Window Functions & Deduplication Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview To find the top-performing or highest-valued entity within each group (the highest-paid employee per department), the canonical approach is using **`row_number()` partitioned by the group attribute (`department`) and ordered by the ranking metric descending (`salary.desc()`)**, followed by filtering for `row_number == 1`.

2. Architecture & Design Principle 1. **Per-Department Parallelism**: - `Window.partitionBy("department")` ensures that records for each department are grouped together on worker nodes. - Salaries are sorted descending within each department. - The employee with the maximum salary receives `row_number = 1`. 2. **Avoiding GroupBy + Join**: - A naive SQL approach finds `max(salary)` via `groupBy("department")` and joins the result back to the original table. This requires two scans and a distributed shuffle join. - `row_number()` achieves the same result in a single pass with zero joins.

3. Step-by-Step Code / Logic Execution 1. Window specification:

   from pyspark.sql.functions import col, row_number
   from pyspark.sql.window import Window

   dept_window = Window.partitionBy("department").orderBy(col("salary").desc())

2. Assign rank and filter:

   top_earners = (
       df.withColumn("rn", row_number().over(dept_window))
         .filter(col("rn") == 1)
         .drop("rn")
   )

3. Output: Exactly one employee per department representing the top earner.

4. Why the Correct Answer Is Optimal Option B is correct because: - It isolates the top record per partition without joining or multi-pass processing. - It returns full employee details (ID, department, salary).

5. Why Alternative Options Fail - Option A (collect()): Incorrect; pulls data to the driver without ranking. - Option C (full join): Incorrect; full outer joins combine disparate tables; they do not rank within partitions. - Option D (cache()): Incorrect; caching stores data in memory without performing transformations.

6. Edge Cases & Production Pitfalls - **Ties in Maximum Salary**: If two employees in "Engineering" both earn $200,000, `row_number()` arbitrarily picks one. If both must be returned, use `dense_rank() == 1`.

7. Performance & Optimization Considerations - In Spark 3.0+, Catalyst recognizes the `row_number() == 1` filter and applies `TakeOrderedAndProject` physical operators to optimize execution.

8. Reliability & Operational Best Practices - Always drop the temporary ranking column before writing to target tables.

9. Real-World Architectural Scenario An enterprise payroll audit system identifies the highest compensation package in each regional business unit to ensure compliance with executive compensation caps.

10. Official Documentation & Learning Resources - Group Ranking: [PySpark Window Functions & Partition Ranking Tutorial](/tutorials/pyspark/df-window-functions)

1. Concept Overview Detecting consecutive events or temporal patterns in event logs requires comparing the timestamp of each event with the timestamp of the immediately preceding event. In PySpark, this is achieved by combining **`lag()`** (to retrieve the previous event's date within a customer window) with **`datediff()`** (to compute the calendar day difference between the current date and the lagged date).

2. Architecture & Design Principle 1. **Window-Based Chronology**: - Events are partitioned by `customer_id` and ordered by `event_date.asc()`. - `lag("event_date", 1)` retrieves the prior event date. 2. **Temporal Difference Metric**: - `datediff(col("event_date"), col("prev_event_date"))` calculates the integer day delta. - Filtering for `datediff == 1` isolates events occurring on consecutive calendar days.

3. Step-by-Step Code / Logic Execution 1. Define window and calculate lag:

   from pyspark.sql.functions import col, datediff, lag
   from pyspark.sql.window import Window

   user_window = Window.partitionBy("customer_id").orderBy("event_date")

   consecutive_events_df = (
       df.withColumn("prev_event_date", lag("event_date", 1).over(user_window))
         .withColumn("days_since_prev", datediff(col("event_date"), col("prev_event_date")))
         .filter(col("days_since_prev") == 1)
   )

2. Execution trace: - Event 1: '2026-03-01' -> Prev = NULL -> Diff = NULL. - Event 2: '2026-03-02' -> Prev = '2026-03-01' -> Diff = 1 day (Consecutive!). - Event 3: '2026-03-05' -> Prev = '2026-03-02' -> Diff = 3 days (Not consecutive).

4. Why the Correct Answer Is Optimal Option D is correct because: - `lag()` accesses the prior row in chronological order. - `datediff()` evaluates the calendar distance between consecutive timestamps.

5. Why Alternative Options Fail - Option A (avg()): Incorrect; `avg()` computes numerical averages. - Option B (sum()): Incorrect; `sum()` computes additions. - Option C (union()): Incorrect; `union()` merges DataFrames.

6. Edge Cases & Production Pitfalls - **Same-Day Multiple Events**: If a customer triggers multiple events on the same day, `datediff()` returns `0`. To detect consecutive events across distinct days, deduplicate by `event_date` before applying `lag()`.

7. Performance & Optimization Considerations - Windowing on `customer_id` scales linearly across executor nodes without expensive self-joins.

8. Reliability & Operational Best Practices - Ensure date strings are properly parsed as `DateType` or `TimestampType` before passing to `datediff()`.

9. Real-World Architectural Scenario A mobile gaming studio awards daily login streak bonuses to players who open the app on consecutive calendar days. Using `lag() + datediff()` powers the automated streak calculation engine for 15 million daily active users.

10. Official Documentation & Learning Resources - Date & Time Functions: [PySpark Dates & Timestamp Manipulation Tutorial](/tutorials/pyspark/pyspark-dates)

1. Concept Overview While PySpark's native `dropDuplicates()` removes duplicates across specified columns, it is **non-deterministic** regarding which row is preserved when duplicate rows differ in other attributes (such as timestamps). To strictly guarantee that the record with the most recent timestamp is retained, you must use **`row_number()` ordered by timestamp descending** within a window partitioned by the entity business key, and filter where `row_number == 1`.

2. Architecture & Design Principle 1. **Deterministic Deduplication**: - `distinct()` or `dropDuplicates(["user_id"])` does not guarantee which row survives. - `Window.partitionBy("user_id").orderBy(col("timestamp").desc())` forces explicit ordering, ensuring that the latest record deterministically receives index 1. 2. **Data Consistency**: In medallion lakehouse architectures (Bronze -> Silver), this pattern is the cornerstone of Silver layer compaction and Change Data Capture (CDC) deduplication.

3. Step-by-Step Code / Logic Execution 1. Construct deterministic deduplication query:

   from pyspark.sql.functions import col, row_number
   from pyspark.sql.window import Window

   dedup_window = (
       Window.partitionBy("user_id")
             .orderBy(col("updated_at").desc())
   )

   clean_df = (
       df.withColumn("rn", row_number().over(dedup_window))
         .filter(col("rn") == 1)
         .drop("rn")
   )

2. Result: For every user_id, only the single most recent record is preserved.

4. Why the Correct Answer Is Optimal Option C is correct because: - Descending timestamp order guarantees that the latest record receives rank 1. - Filtering `rn == 1` retains the newest version deterministically.

5. Why Alternative Options Fail - Option A (distinct()): Incorrect; `distinct()` only removes identical rows; if timestamps differ, `distinct()` keeps all rows! - Option B (count()): Incorrect; returns frequency counts. - Option D (collect()): Incorrect; pulls data to the driver node, risking OOM crashes.

6. Edge Cases & Production Pitfalls - **Identical Timestamps**: If two updates share the exact same timestamp, add a secondary tie-breaker (e.g., `ingestion_id.desc()`) to maintain strict determinism.

7. Performance & Optimization Considerations - In Delta Lake, the `MERGE INTO` statement performs deduplication natively using table index metadata.

8. Reliability & Operational Best Practices - Standardize this deduplication pattern across all Silver ETL ingestion pipelines to prevent data drift.

9. Real-World Architectural Scenario A smart meter IoT platform receives sensor readings that are occasionally resent by edge devices after connection outages. Applying `row_number over timestamp desc` ensures that stale re-transmitted readings are discarded in favor of newer telemetry.

10. Official Documentation & Learning Resources - Deduplication Best Practices: [PySpark Window Functions & Deduplication Tutorial](/tutorials/pyspark/df-window-functions)

1. Concept Overview A **Moving Average** computes the average of a metric over a sliding chronological window. In PySpark, defining the sliding frame requires specifying the boundary offsets using **`rowsBetween()`** (for row-based sliding windows, e.g., `rowsBetween(-6, 0)` for the past 6 rows plus current row) or **`rangeBetween()`** (for numeric/time-based ranges, e.g., past 7 days in seconds).

2. Architecture & Design Principle 1. **Window Frame Specification**: - The window is ordered by `sales_date.asc()`. - The sliding boundary `rowsBetween(-6, 0)` captures a 7-row window (6 preceding days + current day). - Calling `avg("sales_amount").over(w)` calculates the moving average dynamically for each row. 2. **Single-Pass Calculation**: Spark calculates sliding aggregates in a single pass using a sliding accumulator, achieving optimal computational efficiency.

3. Step-by-Step Code / Logic Execution 1. Define moving average window:

   from pyspark.sql.functions import avg, col
   from pyspark.sql.window import Window

   # Assuming one record per day
   seven_day_window = (
       Window.orderBy("sales_date")
             .rowsBetween(-6, Window.currentRow)
   )

   moving_avg_df = df.withColumn(
       "seven_day_mavg", avg("sales_amount").over(seven_day_window)
   )

2. Execution trace: - Days 1 to 6: Average computes over available days up to that point. - Day 7+: Average computes over exactly the 7-day rolling window.

4. Why the Correct Answer Is Optimal Option D is correct because: - `Window Between` (`rowsBetween` / `rangeBetween`) is the native PySpark mechanism for defining sliding window frames. - It dynamically recalculates the average across the moving 7-day boundary.

5. Why Alternative Options Fail - Option A (left join): Incorrect; implementing moving averages via self-joins causes quadratic data explosion. - Option B (distinct()): Incorrect; removes duplicate rows. - Option C (repartition()): Incorrect; changes partition counts without computing sliding metrics.

6. Edge Cases & Production Pitfalls - **Missing Calendar Days**: If days are missing in the dataset (e.g., no weekend sales), `rowsBetween(-6, 0)` averages the last 7 *recorded sales rows*, which might span 10 calendar days! To strictly enforce 7 calendar days, cast the date to unix timestamp and use `rangeBetween(-6 * 86400, 0)`.

7. Performance & Optimization Considerations - Avoid unpartitioned global windows on massive tables; partition by store, region, or product category where possible.

8. Reliability & Operational Best Practices - Validate whether the business defines a 7-day moving average based on calendar days or transaction days.

9. Real-World Architectural Scenario A retail inventory demand forecasting system calculates 7-day, 30-day, and 90-day moving averages of store sales to smooth out weekday/weekend seasonality before training ARIMA forecasting models.

10. Official Documentation & Learning Resources - Window Framing: [PySpark Window Framing & Moving Averages Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview To compare metrics between successive events in an ordered sequence (such as comparing the cost of a current shipment against the cost of the previous shipment on the same route), the **`lag()` window function** is the required solution. `lag("cost", 1)` retrieves the value from the preceding row within an ordered partition, allowing direct subtraction to compute the cost difference.

2. Architecture & Design Principle 1. **Partitioned Offset Lookups**: - Deliveries are partitioned by `route_id` and ordered by `shipment_date.asc()`. - `lag("cost", 1)` accesses the immediately preceding shipment's cost from the executor's memory buffer. 2. **Efficiency vs. Joins**: - `lag()` operates in a single pass over the sorted partition with zero join shuffles.

3. Step-by-Step Code / Logic Execution 1. Window specification:

   from pyspark.sql.functions import col, lag
   from pyspark.sql.window import Window

   route_window = Window.partitionBy("route_id").orderBy("shipment_date")

2. Compute previous cost and delta:

   shipment_comparison_df = (
       df.withColumn("prev_cost", lag("cost", 1).over(route_window))
         .withColumn("cost_diff", col("cost") - col("prev_cost"))
   )

3. Execution trace: - Shipment 1: Cost = $500 -> Prev = NULL -> Diff = NULL. - Shipment 2: Cost = $550 -> Prev = $500 -> Diff = +$50. - Shipment 3: Cost = $520 -> Prev = $550 -> Diff = -$30.

4. Why the Correct Answer Is Optimal Option C is correct because: - `lag()` retrieves prior-row values in sorted partitions. - It enables simple arithmetic calculations of deltas between adjacent events.

5. Why Alternative Options Fail - Option A (count()): Incorrect; computes frequency totals. - Option B (dense_rank()): Incorrect; assigns rank tiers without retrieving historical column values. - Option D (cache()): Incorrect; persists data in memory without transforming columns.

6. Edge Cases & Production Pitfalls - **First Shipment**: The first shipment on each route has no predecessor and evaluates to `NULL`. Use `lag("cost", 1, col("cost"))` or `fillna(0)` if a numerical delta of 0 is preferred.

7. Performance & Optimization Considerations - Windowing partitioned by `route_id` distributes workload evenly across executor cores.

8. Reliability & Operational Best Practices - Always verify date column format and ensure sorting order is deterministic.

9. Real-World Architectural Scenario A maritime shipping logistics network monitors fuel and transit costs across container vessels. Comparing current voyage costs against prior voyages using `lag()` flags anomalous route delays and fuel price spikes.

10. Official Documentation & Learning Resources - Window Functions: [PySpark Window Functions & Offset Operations Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview Detecting multi-day streaks (such as 3 consecutive days of ordering) requires verifying that the interval between the current order date, the previous order date, and the order date two days prior are each exactly 1 calendar day apart. In PySpark, this is achieved by combining a **Window** partitioned by `customer_id` with two **`lag()`** calls (offset 1 and offset 2) and **`datediff()`** comparisons.

2. Architecture & Design Principle 1. **Multi-Offset Window Traversal**: - Order dates are deduplicated per customer and sorted ascending. - `lag("order_date", 1)` retrieves the previous day's order. - `lag("order_date", 2)` retrieves the order from two orders prior. 2. **Consecutive Streak Verification**: - Check 1: `datediff(order_date, prev_date_1) == 1` (Yesterday). - Check 2: `datediff(prev_date_1, prev_date_2) == 1` (Day before yesterday). - When both conditions are True, the customer has achieved a 3-day consecutive order streak!

3. Step-by-Step Code / Logic Execution 1. Deduplicate by customer and date:

   daily_orders = df.select("customer_id", "order_date").distinct()

2. Window specification and streak logic:

   from pyspark.sql.functions import col, datediff, lag
   from pyspark.sql.window import Window

   w = Window.partitionBy("customer_id").orderBy("order_date")

   streak_df = (
       daily_orders
         .withColumn("prev_date_1", lag("order_date", 1).over(w))
         .withColumn("prev_date_2", lag("order_date", 2).over(w))
         .filter(
             (datediff(col("order_date"), col("prev_date_1")) == 1) &
             (datediff(col("prev_date_1"), col("prev_date_2")) == 1)
         )
         .select("customer_id")
         .distinct()
   )

3. Output: Customers who completed at least one 3-day consecutive purchase streak.

4. Why the Correct Answer Is Optimal Option D is correct because: - It solves the streak detection problem natively in distributed memory without self-joins. - Combining `lag()` and `datediff()` across a window provides exact calendar streak validation.

5. Why Alternative Options Fail - Option A (union): Incorrect; concatenates datasets without evaluating sequences. - Option B (broadcast): Incorrect; an optimization technique for small table joins. - Option C (cache()): Incorrect; in-memory storage mechanism.

6. Edge Cases & Production Pitfalls - **Multiple Orders on Same Day**: Without pre-deduplicating dates (`daily_orders.distinct()`), two orders placed on Tuesday would make the gap between Tuesday's second order and Tuesday's first order 0 days, breaking the 1-day check. Always deduplicate dates first.

7. Performance & Optimization Considerations - Deduplicating dates before window evaluation reduces the number of rows processed by the window operator.

8. Reliability & Operational Best Practices - Ensure date columns are formatted as `DateType` so `datediff()` accurately accounts for month boundaries and leap years.

9. Real-World Architectural Scenario A quick-service coffee loyalty app offers bonus rewards to customers who purchase beverages three days in a row. Applying this window streak logic identifies qualifying loyalty members for targeted promotion credits.

10. Official Documentation & Learning Resources - Time-Series Analysis: [PySpark Window Functions & Date Analytics Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview Finding the single top item per group across an enterprise dataset (such as the top-selling product in each category) is solved optimally in PySpark by using **`row_number()` partitioned by the grouping column (`category`) and ordered by sales descending (`sales.desc()`)**, followed by filtering for `row_number == 1`.

2. Architecture & Design Principle 1. **Category Partitioning**: - `Window.partitionBy("category")` groups all products of a given category onto the same executor task. - Products are sorted by total sales descending. - The top product receives `row_number = 1`. 2. **Single Distributed Pass**: This eliminates the need to calculate category maximums in a separate aggregation and join the results back.

3. Step-by-Step Code / Logic Execution 1. Window specification:

   from pyspark.sql.functions import col, row_number
   from pyspark.sql.window import Window

   cat_window = Window.partitionBy("category").orderBy(col("total_sales").desc())

2. Rank and filter:

   top_product_per_category = (
       df.withColumn("rn", row_number().over(cat_window))
         .filter(col("rn") == 1)
         .drop("rn")
   )

3. Output: Exactly one row per product category containing the top-selling product details.

4. Why the Correct Answer Is Optimal Option B is correct because: - It isolates the highest-selling product in each category in a single distributed step. - It returns full product metadata (product ID, product name, supplier, margin).

5. Why Alternative Options Fail - Option A (collect): Incorrect; pulls all records to the driver node, crashing with driver OOM. - Option C (repartition(1)): Incorrect; forces all cluster data into a single partition, destroying distributed parallelism. - Option D (distinct): Incorrect; removes duplicate rows; it cannot perform ranking.

6. Edge Cases & Production Pitfalls - **Tied Sales**: If two products in "Electronics" tie for top sales, `row_number()` picks one deterministically if a secondary sort key is provided (e.g., `orderBy(col("total_sales").desc(), col("margin").desc())`).

7. Performance & Optimization Considerations - Spark's Catalyst optimizer pushes down the `rn == 1` filter to prune non-top records early during execution.

8. Reliability & Operational Best Practices - Verify that sales totals are aggregated per product before applying the window rank.

9. Real-World Architectural Scenario An e-commerce marketplace homepage features the best-selling product for each of 300 catalog categories. Executing `row_number()` partitioned by category refreshes the homepage catalog recommendations every hour.

10. Official Documentation & Learning Resources - Category Ranking: [PySpark Window Functions & Top Products Tutorial](/tutorials/pyspark/df-window-functions)

1. Concept Overview To compare an attribute of the current record against the immediately preceding record in an ordered partition (such as comparing a customer's current order amount against their prior order amount), the **`lag()` window function** is the required construct. By retrieving the previous order amount with `lag("order_amount", 1)`, the query can simply filter for `col("order_amount") > col("prev_order_amount")`.

2. Architecture & Design Principle 1. **Temporal Partitioning**: - Partition by `customer_id`, order by `order_date.asc()`. - `lag("order_amount", 1)` peeks at the preceding row within the partition memory buffer. 2. **Predicate Evaluation**: - The boolean condition `col("order_amount") > col("prev_order_amount")` evaluates for each row. - For a customer's first order, the lagged value is `NULL`, and the filter condition evaluates to `NULL` (excluded).

3. Step-by-Step Code / Logic Execution 1. Define window and calculate lag:

   from pyspark.sql.functions import col, lag
   from pyspark.sql.window import Window

   w = Window.partitionBy("customer_id").orderBy("order_date")

   larger_orders = (
       df.withColumn("prev_amount", lag("order_amount", 1).over(w))
         .filter(col("order_amount") > col("prev_amount"))
   )

2. Execution trace for customer C1: - Order 1: $100 -> Prev = NULL -> Filter: Excluded. - Order 2: $150 -> Prev = $100 -> 150 > 100 is True -> Retained! - Order 3: $120 -> Prev = $150 -> 120 > 150 is False -> Excluded.

4. Why the Correct Answer Is Optimal Option A is correct because: - `lag()` retrieves prior-row values in time-ordered partitions. - It enables direct inequality comparisons between consecutive events.

5. Why Alternative Options Fail - Option B (lead()): Incorrect; `lead()` looks forward to future orders, not preceding orders. - Option C (rank()): Incorrect; `rank()` assigns ordinal numbers, not historical transaction values. - Option D (avg()): Incorrect; `avg()` computes numerical averages.

6. Edge Cases & Production Pitfalls - **First Order Handling**: The first order has no predecessor (`prev_amount = NULL`). Because comparisons with NULL evaluate to false/unknown, the first order is never retained. If the first order should be treated as an increase, use `coalesce(col("prev_amount"), lit(0))`.

7. Performance & Optimization Considerations - Windowing partitioned by `customer_id` scales across cluster executors without cross-network shuffle joins.

8. Reliability & Operational Best Practices - Ensure ordering includes a timestamp or sequential order ID to break ties if multiple orders occur on the same date.

9. Real-World Architectural Scenario A retail growth marketing team identifies "upselling momentum"—customers whose order sizes are steadily growing over time—to invite them into exclusive loyalty tiers.

10. Official Documentation & Learning Resources - Window Functions: [PySpark Window Functions & Offset Analytics](/tutorials/pyspark/df-window-functions)

1. Concept Overview Calculating Month-over-Month (MoM) percentage change requires accessing the balance from the preceding month for the same customer: `((current_balance - prev_balance) / prev_balance) * 100`. In PySpark, retrieving the prior month's balance within a time-ordered customer partition is achieved using the **`lag()` window function**.

2. Architecture & Design Principle 1. **Time-Series Metric Calculation**: - Accounts are partitioned by `account_id` and ordered by `month_date.asc()`. - `lag("closing_balance", 1)` retrieves the prior month's closing balance. 2. **Mathematical Derivation**: - Growth Delta: `col("closing_balance") - col("prev_closing_balance")`. - Percentage Growth: `(col("closing_balance") - col("prev_closing_balance")) / col("prev_closing_balance") * 100`.

3. Step-by-Step Code / Logic Execution 1. Construct window and calculate MoM metrics:

   from pyspark.sql.functions import col, lag, round
   from pyspark.sql.window import Window

   acct_window = Window.partitionBy("account_id").orderBy("month_date")

   mom_df = (
       df.withColumn("prev_balance", lag("closing_balance", 1).over(acct_window))
         .withColumn("mom_change_pct",
                     round(((col("closing_balance") - col("prev_balance")) / col("prev_balance")) * 100, 2))
   )

2. Execution trace for Account A10: - Jan 2026: $10,000 -> Prev = NULL -> MoM = NULL. - Feb 2026: $12,000 -> Prev = $10,000 -> MoM = +20.0%. - Mar 2026: $11,400 -> Prev = $12,000 -> MoM = -5.0%.

4. Why the Correct Answer Is Optimal Option D is correct because: - `lag()` accesses prior period metrics within ordered partitions. - It enables straightforward percentage-change and trend calculations.

5. Why Alternative Options Fail - Option A (collect()): Incorrect; pulls data to the driver node, causing OOM. - Option B (union()): Incorrect; merges datasets vertically. - Option C (broadcast()): Incorrect; a join optimization hint.

6. Edge Cases & Production Pitfalls - **Division by Zero**: If `prev_balance` is 0, the percentage calculation produces `Infinity` or `NULL`. Wrap the denominator in a `when(col("prev_balance") == 0, None)` condition to prevent mathematical errors.

7. Performance & Optimization Considerations - Partitioning by `account_id` ensures that balance histories are evaluated in parallel across worker nodes.

8. Reliability & Operational Best Practices - Cast closing balances to `DecimalType` to prevent floating-point rounding errors in financial reporting.

9. Real-World Architectural Scenario A commercial bank monitors depositor liquidity trends. Generating monthly balance growth percentages with `lag()` alerts risk managers when commercial clients experience sudden double-digit cash drawdowns.

10. Official Documentation & Learning Resources - Financial Time-Series: [PySpark Window Functions & Trend Analysis](/tutorials/pyspark/df-window-functions)

1. Concept Overview Computing a cumulative running total that resets at the beginning of every calendar year requires a window specification partitioned by the calendar year (`year(transaction_date)`) and ordered by the transaction date (`transaction_date.asc()`). Applying **`Window + sum("amount")`** calculates the progressive cumulative sum within each year, resetting automatically when the year boundary advances.

2. Architecture & Design Principle 1. **Year-Partitioned Windowing**: - `Window.partitionBy(year("transaction_date")).orderBy("transaction_date")` isolates each calendar year into its own partition. - For each year, the running sum starts fresh from zero on January 1st and accumulates through December 31st. 2. **Memory Efficiency**: Because the window is partitioned by year, Spark evaluates different years concurrently across executor partitions.

3. Step-by-Step Code / Logic Execution 1. Extract year and define window:

   from pyspark.sql.functions import col, sum, year
   from pyspark.sql.window import Window

   df_with_year = df.withColumn("txn_year", year("transaction_date"))

   ytd_window = (
       Window.partitionBy("txn_year")
             .orderBy("transaction_date")
             .rowsBetween(Window.unboundedPreceding, Window.currentRow)
   )

2. Compute Year-to-Date (YTD) cumulative revenue:

   ytd_df = df_with_year.withColumn("ytd_revenue", sum("amount").over(ytd_window))

3. Execution trace: - 2025-12-31: Total reaches $1,500,000. - 2026-01-01: txn_year changes from 2025 to 2026 -> New partition starts -> YTD resets to January 1st's amount ($10,000).

4. Why the Correct Answer Is Optimal Option B is correct because: - `Window + sum()` partitioned by year naturally resets cumulative sums at year boundaries. - It preserves transaction-level granularity while appending the progressive YTD revenue.

5. Why Alternative Options Fail - Option A (collect()): Incorrect; centralizes data on the driver node. - Option C (distinct()): Incorrect; removes duplicate rows. - Option D (min()): Incorrect; finds minimum values.

6. Edge Cases & Production Pitfalls - **Missing Dates**: If no transactions occur on some days, the cumulative sum simply carries forward the previous total when the next transaction occurs. - **Multiple Transactions on Same Date**: Using explicit `rowsBetween(Window.unboundedPreceding, Window.currentRow)` ensures proper progressive accumulation.

7. Performance & Optimization Considerations - If data spans many years, partitioning by year ensures balanced workload distribution across the cluster.

8. Reliability & Operational Best Practices - Ensure date strings are parsed into `DateType` so the `year()` extraction function evaluates accurately.

9. Real-World Architectural Scenario A corporate finance reporting dashboard displays daily Year-to-Date (YTD) revenue progression against annual budget targets, resetting on January 1st of every fiscal year.

10. Official Documentation & Learning Resources - Cumulative Aggregations: [PySpark Window Functions & YTD Reporting](/tutorials/pyspark/df-window-functions)

1. Concept Overview Detecting rapid fluctuations or spikes in time-series telemetry data requires comparing each reading with the reading immediately preceding it. In PySpark, **`lag("temperature", 1)`** partitioned by sensor ID and ordered by timestamp retrieves the prior reading. The filter condition `col("today_temperature") > col("previous_temperature") * 2` then identifies all anomaly occurrences where temperature doubled between consecutive pings.

2. Architecture & Design Principle 1. **Time-Series Sensor Window**: - Readings are partitioned by `sensor_id` and ordered by `reading_time.asc()`. - For each reading, Spark accesses the previous reading from the partition memory buffer. 2. **Local Memory Evaluation**: `lag()` requires zero network shuffle beyond the initial partition shuffle, making it ideal for high-throughput IoT streams.

3. Step-by-Step Code / Logic Execution 1. Construct window and calculate lag:

   from pyspark.sql.functions import col, lag
   from pyspark.sql.window import Window

   sensor_window = Window.partitionBy("sensor_id").orderBy("reading_time")

   anomalies_df = (
       df.withColumn("prev_temp", lag("temperature", 1).over(sensor_window))
         .filter(col("temperature") > (col("prev_temp") * 2))
   )

2. Execution trace for Sensor S42: - Reading 1: Temp = 25°C -> Prev = NULL -> Filter: Excluded. - Reading 2: Temp = 28°C -> Prev = 25°C -> 28 > 50 is False -> Excluded. - Reading 3: Temp = 62°C -> Prev = 28°C -> 62 > 56 is True -> Anomaly Flagged!

4. Why the Correct Answer Is Optimal Option A is correct because: - `lag()` accesses prior temporal values within partitioned sensor streams. - It enables instantaneous threshold and multiplier evaluations.

5. Why Alternative Options Fail - Option B (count()): Incorrect; returns frequency totals. - Option C (union()): Incorrect; merges datasets vertically. - Option D (cache()): Incorrect; caches data without transforming columns.

6. Edge Cases & Production Pitfalls - **Negative Temperatures**: If temperatures can be negative (e.g., in industrial freezers), multiplying by 2 reverses inequalities. For robust anomaly detection across all temperature ranges, evaluate absolute changes: `abs(col("temperature") - col("prev_temp")) > threshold`.

7. Performance & Optimization Considerations - In structured streaming pipelines, watermarking and stateful streaming operations (`mapGroupsWithState`) can be used for real-time anomaly detection.

8. Reliability & Operational Best Practices - Filter out `prev_temp.isNull()` explicitly to ensure only valid pairwise comparisons are evaluated.

9. Real-World Architectural Scenario A chemical refinery monitors catalytic reactor temperatures across 5,000 thermal probes. Spikes exceeding 2x prior readings trigger emergency cooling alerts via automated IoT streaming pipelines.

10. Official Documentation & Learning Resources - Window Operations: [PySpark Window Functions & Sensor Analytics Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview A **Left Anti Join** (`how="left_anti"` or `how="anti"`) returns rows from the left DataFrame (`customers_df`) that have **no matching keys** in the right DataFrame (`orders_df`). It directly implements the relational set difference: `Customers - Orders`. Because it only outputs unmatched left rows and never projects right-table columns, it is both syntactically concise and highly optimized in Apache Spark.

2. Architecture & Design Principle 1. **Relational Semantics**: - Traditional approach: Left outer join followed by `filter(orders.customer_id.isNull())`. This requires shuffling and constructing joined rows before filtering. - Anti Join: Spark's Catalyst optimizer stops scanning the right table for a key the moment a match is detected (short-circuiting), emitting only unmatched left keys without materializing null columns. 2. **Execution Strategy**: Implemented via `BroadcastHashJoinExec` (if right table is small) or `SortMergeJoinExec` (for large tables).

3. Step-by-Step Code / Logic Execution 1. Execute left anti join:

   inactive_customers = customers_df.join(
       orders_df,
       on="customer_id",
       how="left_anti"
   )

2. Execution trace: - Customer C1 has orders -> Matched on right -> Excluded. - Customer C2 has orders -> Matched on right -> Excluded. - Customer C3 has NO orders -> Not matched on right -> Emitted! 3. Output: DataFrame containing only customers with zero orders.

4. Why the Correct Answer Is Optimal Option B is correct because: - Left Anti Join directly expresses the requirement "find records in A that do not exist in B". - It avoids unnecessary column projection and provides superior execution speed compared to outer joins with null filters.

5. Why Alternative Options Fail - Option A (Right Join): Incorrect; returns all orders and matching customers. - Option C (Inner Join): Incorrect; returns customers who *did* place orders. - Option D (Cross Join): Incorrect; produces Cartesian product of all customers and all orders.

6. Edge Cases & Production Pitfalls - **Null Keys**: In standard SQL `NOT IN` subqueries, a single `NULL` in the subquery can cause the entire query to return zero rows. PySpark's `left_anti` join safely avoids this trap, correctly returning left rows where the key has no match on the right.

7. Performance & Optimization Considerations - If `orders_df` has only `customer_id` projected before joining (`orders_df.select("customer_id").distinct()`), Spark reduces shuffle network payload significantly.

8. Reliability & Operational Best Practices - Use `left_semi` to find records that *do* have matches, and `left_anti` to find records that *do not* have matches.

9. Real-World Architectural Scenario A marketing retention campaign identifies registered platform users who signed up over 30 days ago but never placed an order, delivering automated welcome discount codes to stimulate first purchases.

10. Official Documentation & Learning Resources - Join Optimization: [PySpark Joins & Anti Join Guide](/tutorials/pyspark/df-joins)

1. Concept Overview **Window Functions** are the fundamental foundation of modern advanced analytics in SQL and Apache Spark. They allow calculations across a defined set of rows (a "window") related to the current row without collapsing the dataset into a single grouped row. Window functions enable running totals, moving averages, cumulative percentages, lead/lag period comparisons, and top-N ranking to be executed efficiently in distributed memory.

2. Architecture & Design Principle 1. **Row-Level Granularity Preservation**: - Traditional `groupBy()` aggregates collapse multiple rows into one. - Window functions append aggregated or positional metrics to every individual row while keeping the original row structure intact. 2. **Avoiding Self-Join Disasters**: - Before window functions, computing a running total or prior-period delta required joining a table to itself on inequalities (`t1.date >= t2.date`), resulting in Cartesian quadratic complexity (`O(N^2)`). - Window functions evaluate partitions in linear time (`O(N log N)` sort followed by `O(N)` scan), scaling to billions of records.

3. Step-by-Step Code / Logic Execution 1. Architectural pillars of PySpark Windowing: - **Partitioning**: `Window.partitionBy("key")` distributes calculation domains across executor nodes. - **Ordering**: `Window.orderBy("sort_col")` establishes the chronological or numeric sequence. - **Framing**: `rowsBetween` / `rangeBetween` bounds the sliding evaluation scope. 2. Core window function families: - **Ranking**: `row_number()`, `rank()`, `dense_rank()`, `ntile()`. - **Analytic Offsets**: `lag()`, `lead()`, `first()`, `last()`. - **Aggregate Over Window**: `sum().over()`, `avg().over()`, `max().over()`.

4. Why the Correct Answer Is Optimal Option C is correct because: - Window functions are universally recognized as the essential PySpark capability for advanced analytics, time-series metrics, and group-level rankings. - They eliminate self-joins and ensure linear distributed performance.

5. Why Alternative Options Fail - Option A (print()): Incorrect; a Python built-in for debugging text output to the console. - Option B (show()): Incorrect; a DataFrame action that prints the first 20 rows to the console. - Option D (limit()): Incorrect; restricts the number of returned rows.

6. Edge Cases & Production Pitfalls - **Global Unpartitioned Windows**: Never execute `Window.orderBy("col")` on a multi-billion row dataset without a `partitionBy()` clause unless absolutely necessary, as it moves all data to a single partition on one executor core.

7. Performance & Optimization Considerations - In Spark 3.x, Adaptive Query Execution (AQE) optimizes window shuffles and dynamically coalesces shuffle partitions.

8. Reliability & Operational Best Practices - Explicitly declare window frame bounds (`rowsBetween`) in production pipelines to avoid subtle bugs caused by default `RANGE` frame behavior on duplicate values.

9. Real-World Architectural Scenario An enterprise financial analytics platform computes real-time portfolio risk metrics, moving volatility bands, and client rank tiers across 50 million accounts. Window functions allow these complex calculations to execute in a single distributed Spark job in under 2 minutes.

10. Official Documentation & Learning Resources - Advanced Analytics: [PySpark DataFrame Window Functions Tutorial](/tutorials/pyspark/df-window-functions)