1. Core Concept In relational algebra and distributed query engines, isolating the single latest or earliest event per entity requires assigning a sequential integer to rows within partitioned groups based on a chronological order. In PySpark, **`row_number()`** is the standard window function used to number rows deterministically from 1 to $N$ per partition.

2. Detailed Breakdown In PySpark: - Syntax:

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

  windowSpec = Window.partitionBy("customer_id").orderBy(desc("order_date"))
  df.withColumn("rn", row_number().over(windowSpec)).filter("rn = 1")

• Catalyst & Physical Plan: - Spark plans a WindowExec physical operator. - Rows are shuffled across cluster nodes partitioned by hash(customer_id) % shuffle.partitions using an Exchange hashpartitioning. - Within each partition, rows are sorted in descending order by order_date via Sort [customer_id ASC, order_date DESC]. - row_number() maintains an internal counter that increments by 1 for every record in the partition, resetting to 1 whenever the customer_id changes. - Filtering for rn = 1 isolates exactly one newest record per customer.

3. Why the Correct Answer is Correct The interviewer requirement specifies: *"the latest order for every customer"* where multiple orders exist. - Grouping by customer and ordering by order date descending allows `row_number()` to assign rank 1 to the most recent order. - Filtering by `row_number == 1` returns the exact latest record for every customer. - Therefore, option A is the standard and correct function.

4. Why Each Distractor is Incorrect - **B (count()):** Computes the total number of orders placed by the customer; it cannot identify or extract the latest order record. - **C (distinct()):** Deduplicates identical rows across all columns; it does not sort by timestamp or isolate the newest record. - **D (sum()):** Computes the arithmetic total of numeric values (e.g. order totals); it has no temporal ordering or ranking capability.

5. Code Example

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

spark = SparkSession.builder.appName("LatestOrderDemo").getOrCreate()

orders = [
    (101, "ORD_01", "2023-01-15", 50.0),
    (101, "ORD_02", "2023-03-20", 120.0), # Latest for 101
    (102, "ORD_03", "2023-02-10", 80.0),  # Only order for 102
    (103, "ORD_04", "2023-01-05", 210.0),
    (103, "ORD_05", "2023-04-12", 45.0)   # Latest for 103
]
df = spark.createDataFrame(orders, ["customer_id", "order_id", "order_date", "amount"])

# Define window partitioned by customer, sorted by date descending
window_spec = Window.partitionBy("customer_id").orderBy(desc("order_date"))

latest_orders = (
    df.withColumn("rn", row_number().over(window_spec))
      .filter(col("rn") == 1)
      .drop("rn")
)
latest_orders.show()
# +-----------+--------+----------+------+
# |customer_id|order_id|order_date|amount|
# +-----------+--------+----------+------+
# |        101|  ORD_02|2023-03-20| 120.0|
# |        102|  ORD_03|2023-02-10|  80.0|
# |        103|  ORD_05|2023-04-12|  45.0|
# +-----------+--------+----------+------+

6. Key Takeaway To extract the latest, earliest, or top-1 record per entity, partition by entity ID, order descending by timestamp, and filter for `row_number() == 1`.

7. Production Best Practice If timestamps can tie (two orders placed at the exact same second), `row_number()` will non-deterministically pick one unless a secondary tie-breaker column (such as `order_id`) is added to the `orderBy()` clause: `Window.partitionBy("customer_id").orderBy(desc("order_date"), desc("order_id"))`.

8. Real-World Engineering Scenario In Slowly Changing Dimension Type 1 (SCD1) or CDC (Change Data Capture) ingestion pipelines, upstream database replication logs emit multiple inserts, updates, and deletes for the same primary key. Data engineers deduplicate the stream by partitioning by primary key, ordering by CDC timestamp descending, and filtering for `row_number() == 1` to retain the current state.

9. Common Pitfall / Anti-Pattern Attempting to get the latest record by doing `df.groupBy("customer_id").agg(max("order_date"))` and then inner-joining back to the raw table. This requires two full cluster shuffles and an expensive join, whereas `row_number()` achieves the same result in a single shuffle pass.

10. Tutorial Reference Master windowing and ranking techniques in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In SQL and analytical ranking, when identical values occur, ranking functions handle ties differently: - **`rank()`**: Assigns identical ranks to tied values and **skips subsequent rank numbers** according to the number of tied rows (e.g., 1, 2, 2, 4). - **`dense_rank()`**: Assigns identical ranks to tied values but **does not skip** numbers (e.g., 1, 2, 2, 3). - **`row_number()`**: Assigns strictly consecutive integers without ties (e.g., 1, 2, 3, 4).

2. Detailed Breakdown In PySpark: - Function: `pyspark.sql.functions.rank()`. - Mechanism: - Spark evaluates the order specification within each partition. - When comparing the current row's ordering values against the previous row: - If the values are identical, `rank()` assigns the same rank as the previous row. - If the values differ, `rank()` assigns the rank equal to the current 1-based row index in the partition. - For example, if two employees tie for 2nd place, both receive rank 2. The next employee is the 4th row evaluated, so `rank()` assigns rank 4 (skipping 3).

3. Why the Correct Answer is Correct The prompt explicitly defines: 1. *"If two employees have the same salary, they should receive the same rank."* 2. *"The next rank should be skipped."* This gap-producing ranking behavior is the textbook definition and exact functionality of the **`rank()`** window function. Therefore, option B is correct.

4. Why Each Distractor is Incorrect - **A (dense_rank()):** Assigns ties the same rank, but does NOT skip subsequent ranks (ranks would be 1, 2, 2, 3 rather than 1, 2, 2, 4), directly violating the requirement to skip ranks. - **C (row_number()):** Assigns unique sequential numbers (1, 2, 3, 4), never assigning ties the same rank. - **D (lag()):** Accesses data from a preceding row at a specified physical offset; it does not compute ranks.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import rank, dense_rank, row_number, desc

spark = SparkSession.builder.appName("RankVsDenseRank").getOrCreate()

employees = [
    ("Alice", "Finance", 90000),
    ("Bob", "Finance", 80000),
    ("Charlie", "Finance", 80000),  # Tied with Bob
    ("David", "Finance", 70000)
]
df = spark.createDataFrame(employees, ["name", "department", "salary"])

window_spec = Window.partitionBy("department").orderBy(desc("salary"))

ranked_df = df.withColumn("rank", rank().over(window_spec)) \
              .withColumn("dense_rank", dense_rank().over(window_spec)) \
              .withColumn("row_number", row_number().over(window_spec))

ranked_df.select("name", "salary", "rank", "dense_rank", "row_number").show()
# +-------+------+----+----------+----------+
# |   name|salary|rank|dense_rank|row_number|
# +-------+------+----+----------+----------+
# |  Alice| 90000|   1|         1|         1|
# |    Bob| 80000|   2|         2|         2|
# |Charlie| 80000|   2|         2|         3|
# |  David| 70000|   4|         3|         4|
# +-------+------+----+----------+----------+
# Notice David receives rank 4 (rank 3 is skipped)!

6. Key Takeaway The `rank()` function gives tied records the same rank value and skips subsequent rank numbers corresponding to the number of ties.

7. Production Best Practice Choose carefully between `rank()` and `dense_rank()` when implementing business rules like "Top 3 highest paid employees." If there is a tie for 2nd place, `rank() <= 3` returns Alice, Bob, and Charlie (3 people, ranks 1, 2, 2). If David is rank 4, he is excluded. With `dense_rank() <= 3`, David would also be included (ranks 1, 2, 2, 3). Ensure the business specifies whether ties consume rank slots.

8. Real-World Engineering Scenario In sports tournament analytics (e.g. Olympic leaderboards or golf tournaments), when two competitors tie for the silver medal (rank 2), both receive silver, and no bronze medal is awarded—the next competitor receives 4th place. PySpark pipelines computing official tournament standings use `rank()`.

9. Common Pitfall / Anti-Pattern Omitting the `orderBy()` clause in the window specification when using `rank()`. PySpark requires an explicit ordering specification for ranking functions; calling `rank().over(Window.partitionBy("dept"))` without `orderBy` throws an `AnalysisException`.

10. Tutorial Reference Understand ranking semantics in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In analytical window operations, **`dense_rank()`** computes rankings for ordered rows within a partition such that identical values receive identical rank values, and subsequent ranks **advance consecutively without gaps** (e.g., 1, 2, 2, 3, 4).

2. Detailed Breakdown In PySpark: - Function: `pyspark.sql.functions.dense_rank()`. - Mechanism: - Spark sorts records within each partition according to the `orderBy` clause. - An internal rank counter begins at 1. - As Spark iterates through rows: - If the sorting keys of the current row match the previous row, the counter does not increment; the same rank is assigned. - If the sorting keys change, the counter increments by exactly **1**, regardless of how many rows shared the previous rank. - This guarantees a "dense" sequence of integers ($1, 2, 3, \dots, K$) with zero numerical gaps.

3. Why the Correct Answer is Correct The requirement explicitly states: *"rankings without gaps even when values tie"*. - Handling ties by assigning identical numbers while avoiding gaps is the definitive characteristic of **`dense_rank()`**. - Therefore, option C is the correct function.

4. Why Each Distractor is Incorrect - **A (row_number()):** Generates unique numbers (1, 2, 3, 4) and does not give tied records the same rank. - **B (rank()):** Leaves gaps in the ranking sequence whenever ties occur (e.g. 1, 2, 2, 4), violating "without gaps". - **D (count()):** Aggregates total frequencies; it does not calculate relative ordinal ranks.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import dense_rank, desc

spark = SparkSession.builder.appName("DenseRankDemo").getOrCreate()

plans = [
    ("Cust_A", 100),
    ("Cust_B", 85),
    ("Cust_C", 85),  # Tie
    ("Cust_D", 70)
]
df = spark.createDataFrame(plans, ["customer_id", "loyalty_score"])

window_spec = Window.orderBy(desc("loyalty_score"))

# dense_rank produces consecutive integers 1, 2, 2, 3
ranked = df.withColumn("dense_rank", dense_rank().over(window_spec))
ranked.show()
# +-----------+-------------+----------+
# |customer_id|loyalty_score|dense_rank|
# +-----------+-------------+----------+
# |     Cust_A|          100|         1|
# |     Cust_B|           85|         2|
# |     Cust_C|           85|         2|
# |     Cust_D|           70|         3|
# +-----------+-------------+----------+

6. Key Takeaway Use `dense_rank()` when ranking ordered groups where tied entities share the same rank and subsequent ranks must follow consecutively without skipping integers.

7. Production Best Practice When filtering for "Top N distinct value tiers" (e.g., the top 3 highest salary bands or top 3 customer loyalty tiers), always use `dense_rank() <= N`. Using `rank() <= N` can truncate results prematurely if multiple employees tie in higher tiers.

8. Real-World Engineering Scenario A telecom loyalty engine tiers subscribers into Platinum, Gold, and Silver levels based on annual network spend. Subscribers with identical spending belong to the same tier. Using `dense_rank()`, rank 1 maps to Platinum, rank 2 to Gold, and rank 3 to Silver, ensuring all three award tiers are always populated regardless of ties in rank 1 or 2.

9. Common Pitfall / Anti-Pattern Assuming `dense_rank()` returns the exact number of top entities. If 10 customers all share the highest score (rank 1), filtering `dense_rank() <= 2` will return all 10 customers from rank 1 plus everyone in rank 2. If you need strictly $N$ rows, use `row_number()` instead.

10. Tutorial Reference Read about window function ranking in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In time-series analytics and sequential reporting, accessing values from a **preceding row** (a prior time step) relative to the current row without performing a self-join is accomplished using the **`lag()`** window function.

2. Detailed Breakdown In PySpark: - Syntax: `pyspark.sql.functions.lag(col, offset=1, default=None)`. - Parameters: - `col`: The column whose previous value should be fetched. - `offset`: Number of rows backward to look (default is 1). - `default`: Fallback value returned when the offset falls outside the partition boundary (default is `None` / `NULL`). - Physical Execution: - Inside the `WindowExec` operator, Spark sorts the records chronologically (`orderBy("date")`). - As Spark processes row $i$, it maintains a sliding buffer of preceding rows in memory. - It retrieves the value from row $i - 1$ and emits it alongside row $i$'s attributes. - This provides $O(1)$ access time per row during partition iteration.

3. Why the Correct Answer is Correct The requirement states: *"compare today's patient count with yesterday's patient count"*. - "Yesterday" is the immediately preceding record in a daily sorted sequence. - `lag("patient_count", 1)` looks backward by 1 row to fetch yesterday's count into today's row. - Subtracting `lag("patient_count")` from `patient_count` computes the day-over-day delta. - Therefore, option D is the correct function.

4. Why Each Distractor is Incorrect - **A (lead()):** Looks forward into subsequent rows (tomorrow's count), not backward to yesterday. - **B (rank()):** Calculates ordinal ranking positions; it cannot retrieve previous data values. - **C (distinct()):** Removes duplicate rows; it does not perform temporal comparisons.

5. Code Example

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

spark = SparkSession.builder.appName("PatientLagDemo").getOrCreate()

patient_data = [
    ("Hospital_A", "2023-10-01", 120),
    ("Hospital_A", "2023-10-02", 135),
    ("Hospital_A", "2023-10-03", 128)
]
df = spark.createDataFrame(patient_data, ["hospital", "date", "patient_count"])

# Define chronological window
window_spec = Window.partitionBy("hospital").orderBy("date")

# Access yesterday's patient count and compute daily delta
result = (
    df.withColumn("prev_day_count", lag("patient_count", 1).over(window_spec))
      .withColumn("day_over_day_change", col("patient_count") - col("prev_day_count"))
)
result.show()
# +----------+----------+-------------+--------------+------------------+
# |  hospital|      date|patient_count|prev_day_count|day_over_day_change|
# +----------+----------+-------------+--------------+------------------+
# |Hospital_A|2023-10-01|          120|          NULL|              NULL|
# |Hospital_A|2023-10-02|          135|           120|                15|
# |Hospital_A|2023-10-03|          128|           135|                -7|
# +----------+----------+-------------+--------------+------------------+

6. Key Takeaway The `lag()` window function retrieves values from preceding rows within an ordered partition, enabling day-over-day, month-over-month, and sequential delta calculations.

7. Production Best Practice Ensure there are no missing dates or gaps in your time series before applying `lag(col, 1)`. If a hospital has no record on Oct 2nd, `lag(1)` on Oct 3rd will fetch Oct 1st, potentially misrepresenting a 2-day change as a 1-day change. If strict calendar day lag is needed, join against a complete date dimension table or use range-based windows.

8. Real-World Engineering Scenario In hospital bed management systems during public health emergencies, state health departments track ICU capacity trends. Daily ETL jobs calculate the 24-hour delta in occupied ventilators by subtracting `lag("occupied_ventilators")` to trigger surge protocol alerts.

9. Common Pitfall / Anti-Pattern Omitting `partitionBy()` when multiple hospitals or patients exist in the dataset. If you order only by date without partitioning by hospital, the first day of Hospital B will lag into the last day of Hospital A, corrupting the metrics with cross-entity data leakage!

10. Tutorial Reference Explore lead and lag patterns in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In sequential and temporal data processing, accessing values from a **future or subsequent row** relative to the current position within an ordered window partition is performed using the **`lead()`** function.

2. Detailed Breakdown In PySpark: - Syntax: `pyspark.sql.functions.lead(col, offset=1, default=None)`. - Parameters: - `col`: Target column from which future values will be retrieved. - `offset`: Number of rows forward to look (default 1). - `default`: Optional default value when the offset exceeds the final row of the partition (default is `NULL`). - Mechanism: - Spark's `WindowExec` buffers the sorted partition stream. - When outputting row $i$, Spark peeks forward to row $i + 1$ and projects its value into the current row. - On the final row of each partition, `lead()` returns `NULL` (or the configured default) because no subsequent record exists.

3. Why the Correct Answer is Correct The business requirement states: *"view the next scheduled delivery date for each shipment"*. - "Next" refers to looking forward in chronological order. - `lead("delivery_date", 1)` looks ahead by 1 row to fetch the upcoming scheduled delivery date. - Therefore, option A is the correct function.

4. Why Each Distractor is Incorrect - **B (lag()):** Looks backward to previous rows (past deliveries), not forward to upcoming deliveries. - **C (row_number()):** Generates an integer rank (1, 2, 3); it cannot extract date values from future rows. - **D (avg()):** Calculates arithmetic means; it does not access subsequent rows.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import lead

spark = SparkSession.builder.appName("LeadDeliveryDemo").getOrCreate()

deliveries = [
    ("Truck_1", "2023-11-01"),
    ("Truck_1", "2023-11-05"),
    ("Truck_1", "2023-11-12"),
    ("Truck_2", "2023-11-02")
]
df = spark.createDataFrame(deliveries, ["truck_id", "delivery_date"])

window_spec = Window.partitionBy("truck_id").orderBy("delivery_date")

result = df.withColumn("next_delivery_date", lead("delivery_date", 1).over(window_spec))
result.show()
# +--------+-------------+------------------+
# |truck_id|delivery_date|next_delivery_date|
# +--------+-------------+------------------+
# | Truck_1|   2023-11-01|        2023-11-05|
# | Truck_1|   2023-11-05|        2023-11-12|
# | Truck_1|   2023-11-12|              NULL|
# | Truck_2|   2023-11-02|              NULL|
# +--------+-------------+------------------+

6. Key Takeaway The `lead()` window function retrieves values from subsequent rows within an ordered partition, allowing pipelines to project upcoming events and calculate lead times.

7. Production Best Practice Always handle the final row's `NULL` value when performing calculations with `lead()`. For example, calculating transit duration (`datediff(lead("delivery_date").over(w), col("delivery_date"))`) will return `NULL` on the last delivery. Use `coalesce()` if a fallback date or default duration is required.

8. Real-World Engineering Scenario In fleet management and freight logistics, dispatchers compute vehicle idle time between assignments. PySpark jobs order all trips for each driver chronologically and use `lead("pickup_time") - col("dropoff_time")` to measure downtime and optimize routing schedules.

9. Common Pitfall / Anti-Pattern Confusing `lead()` and `lag()`. Remember: `lag()` looks *behind* (past events), while `lead()` looks *ahead* (future events). Reversing them reverses the timeline, resulting in negative durations or incorrect causal modeling.

10. Tutorial Reference Read about `lead()` and `lag()` implementation in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept Finding the "most recent record per entity" is a classic deduplication and latest-state extraction problem. In PySpark, **`row_number()`** combined with a window partition ordered descending by timestamp provides a deterministic, single-row-per-entity guarantee.

2. Detailed Breakdown When querying the most recent transaction per customer: - If a customer has multiple transactions:

  w = Window.partitionBy("customer_id").orderBy(col("txn_timestamp").desc())
  df.withColumn("rn", row_number().over(w)).filter(col("rn") == 1)

• Why not max("txn_timestamp")? - Calling df.groupBy("customer_id").agg(max("txn_timestamp")) only gives you the timestamp itself! - It discards all other transaction columns (transaction amount, merchant name, payment method, location). - To get the *entire transaction record* without joining the aggregated timestamp back to the base table, row_number() over a window is the industry-standard design pattern.
• Why row_number() over rank() or dense_rank()? - If two transactions share the exact same microsecond timestamp, rank() and dense_rank() assign 1 to both, returning duplicate transactions for that customer. - row_number() guarantees strictly one row per customer.

3. Why the Correct Answer is Correct - The requirement is to find the most recent transaction (the full row) per customer. - `row_number()` over a partition ordered by transaction timestamp descending assigns `1` to the newest transaction and guarantees exactly one record per customer when filtered to `rn == 1`. - Therefore, option B is the standard and correct function.

4. Why Each Distractor is Incorrect - **A (dense_rank()):** If a customer executes two transactions with identical timestamps, both receive rank 1, failing to return a single unique transaction per customer. - **C (sum()):** Aggregates numeric values; it cannot identify or select the most recent record. - **D (max()):** While `max()` finds the highest timestamp, it cannot easily return the other columns of that transaction record in a single operation without an expensive join.

5. Code Example

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

spark = SparkSession.builder.appName("RecentTxnDemo").getOrCreate()

txns = [
    ("Cust_1", "TX_01", "2023-08-01 10:00:00", 25.0),
    ("Cust_1", "TX_02", "2023-08-01 14:30:00", 110.0), # Most recent for Cust_1
    ("Cust_2", "TX_03", "2023-08-02 09:15:00", 40.0)
]
df = spark.createDataFrame(txns, ["customer_id", "txn_id", "txn_time", "amount"])

# Window to isolate top-1 recent transaction
w = Window.partitionBy("customer_id").orderBy(desc("txn_time"))

latest_txn = df.withColumn("rn", row_number().over(w)).filter(col("rn") == 1).drop("rn")
latest_txn.show()
# +-----------+------+-------------------+------+
# |customer_id|txn_id|           txn_time|amount|
# +-----------+------+-------------------+------+
# |     Cust_1| TX_02|2023-08-01 14:30:00| 110.0|
# |     Cust_2| TX_03|2023-08-02 09:15:00|  40.0|
# +-----------+------+-------------------+------+

6. Key Takeaway `row_number()` filtered to `1` over a descending timestamp window is the canonical pattern to retrieve the entire most recent record per entity in PySpark.

7. Production Best Practice To ensure deterministic output in production when timestamps tie, include a secondary tie-breaker in the `orderBy` clause: `orderBy(desc("txn_time"), desc("txn_id"))`. This ensures reproducible results across pipeline reruns.

8. Real-World Engineering Scenario In banking core systems, credit risk scoring models require the user's latest account balance and latest transaction status. A nightly batch job runs `row_number()` over all transaction logs to extract the latest transaction per account and publishes it to the feature store.

9. Common Pitfall / Anti-Pattern Using `rank()` instead of `row_number()` for deduplication. If duplicate timestamps exist in upstream data, `filter(rank == 1)` leaks multiple records for the same customer into downstream gold tables, causing primary key constraint violations.

10. Tutorial Reference Master deduplication and latest-row extraction in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept A **running balance** (or cumulative sum) requires computing the progressive accumulation of values row-by-row in chronological sequence. In PySpark, this is accomplished by applying **`sum()` over an ordered Window specification**.

2. Detailed Breakdown In PySpark: - When an aggregate function like `sum(col)` is invoked over an ordered window:

  w = Window.partitionBy("account_id").orderBy("txn_date")
  df.withColumn("running_balance", sum("amount").over(w))

• Default Frame Specification: - In Spark SQL, when an orderBy is provided without an explicit frame clause, Spark automatically defaults to: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW - This means for every row, Spark sums all transactions from the start of the account's history up to the current row's timestamp.
• For optimal performance and exact row semantics, specifying rowsBetween(Window.unboundedPreceding, Window.currentRow) is recommended.

3. Why the Correct Answer is Correct The requirement states: *"a running account balance after every transaction"*. - Each transaction row must remain in the DataFrame while displaying the cumulative sum of all transactions up to that point. - Applying **`sum() over Window`** calculates this cumulative balance while preserving every individual transaction row. - Therefore, option C is the correct pattern.

4. Why Each Distractor is Incorrect - **A (dropDuplicates()):** Removes duplicate rows; it does not compute balances or sums. - **B (collect()):** Pulls all DataFrame records across the cluster into driver Python memory as a list, which crashes driver memory and performs no mathematical calculation. - **D (distinct()):** Deduplicates rows; it cannot calculate cumulative financial metrics.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import sum as _sum

spark = SparkSession.builder.appName("RunningBalanceDemo").getOrCreate()

txns = [
    ("ACC_001", "2023-09-01", 1000.0), # Deposit
    ("ACC_001", "2023-09-03", -200.0), # Withdrawal
    ("ACC_001", "2023-09-05", 500.0),  # Deposit
    ("ACC_002", "2023-09-01", 300.0)
]
df = spark.createDataFrame(txns, ["account_id", "txn_date", "amount"])

# Running balance window
w = Window.partitionBy("account_id").orderBy("txn_date") \
          .rowsBetween(Window.unboundedPreceding, Window.currentRow)

running_df = df.withColumn("running_balance", _sum("amount").over(w))
running_df.show()
# +----------+----------+------+---------------+
# |account_id|  txn_date|amount|running_balance|
# +----------+----------+------+---------------+
# |   ACC_001|2023-09-01|1000.0|         1000.0|
# |   ACC_001|2023-09-03|-200.0|          800.0|
# |   ACC_001|2023-09-05| 500.0|         1300.0|
# |   ACC_002|2023-09-01| 300.0|          300.0|
# +----------+----------+------+---------------+

6. Key Takeaway Applying `sum()` over an ordered window partitioned by account computes a cumulative running total while preserving individual transaction detail.

7. Production Best Practice Always be explicit with frame boundaries! If multiple transactions have identical timestamps and you use the default `RANGE BETWEEN`, Spark sums all transactions sharing that timestamp simultaneously. To guarantee strictly progressive accumulation transaction-by-transaction, explicitly declare `rowsBetween(Window.unboundedPreceding, Window.currentRow)`.

8. Real-World Engineering Scenario In core banking ledgers and financial auditing pipelines, every debit and credit transaction must output the verified post-transaction ledger balance. This ensures customer statements and regulatory balance histories match actual accounting records.

9. Common Pitfall / Anti-Pattern Omitting `partitionBy("account_id")`. If you only specify `Window.orderBy("txn_date")`, Spark routes **all accounts in the entire bank** into a single partition on a single executor to compute a global running total, causing massive executor Out-Of-Memory (OOM) failures.

10. Tutorial Reference Learn how to build cumulative running totals in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In business intelligence and revenue reporting, tracking performance trajectory over a reporting period (such as Year-to-Date or Month-to-Date sales) requires computing a cumulative sum that progresses day by day. In PySpark, this is known as a **Running Total Window**.

2. Detailed Breakdown How a Running Total Window operates: - Specification:

  w = Window.partitionBy("region").orderBy("sale_date") \
            .rowsBetween(Window.unboundedPreceding, Window.currentRow)
  df.withColumn("cumulative_sales", sum("daily_sales").over(w))

• Execution Mechanics: - Rows are partitioned by business unit or region and sorted chronologically by date. - The frame definition unboundedPreceding to currentRow instructs Spark: for each day, sum the sales of that day with all prior days in the partition. - Unlike groupBy(), which collapses multiple daily rows into a single aggregate row, a window function preserves every individual daily record while adding the cumulative total as a new column.

3. Why the Correct Answer is Correct The requirement explicitly states: *"cumulative sales as each day passes"*. - "As each day passes" indicates that daily records must be preserved in chronological sequence. - "Cumulative sales" indicates a progressive running sum. - A **Running Total Window** is designed specifically to compute this metric. - Therefore, option D is the correct approach.

4. Why Each Distractor is Incorrect - **A (Union):** Concatenates two DataFrames together vertically; it cannot compute cumulative temporal totals. - **B (Left Join):** Matches records between two tables; without a window, a self-join to compute running totals requires an inefficient $\frac{N(N+1)}{2}$ row explosion. - **C (Distinct):** Removes duplicate rows; it performs no mathematical calculations.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import sum as _sum

spark = SparkSession.builder.appName("CumulativeSalesDemo").getOrCreate()

sales_data = [
    ("2023-05-01", 1000.0),
    ("2023-05-02", 1500.0),
    ("2023-05-03", 2000.0),
    ("2023-05-04", 1200.0)
]
df = spark.createDataFrame(sales_data, ["sale_date", "daily_sales"])

# Window sorted by sale_date from beginning of time to current row
cum_window = Window.orderBy("sale_date").rowsBetween(Window.unboundedPreceding, Window.currentRow)

cum_df = df.withColumn("cumulative_sales", _sum("daily_sales").over(cum_window))
cum_df.show()
# +----------+-----------+----------------+
# | sale_date|daily_sales|cumulative_sales|
# +----------+-----------+----------------+
# |2023-05-01|     1000.0|          1000.0|
# |2023-05-02|     1500.0|          2500.0|
# |2023-05-03|     2000.0|          4500.0|
# |2023-05-04|     1200.0|          5700.0|
# +----------+-----------+----------------+

6. Key Takeaway A Running Total Window calculates progressive cumulative metrics across chronological records without collapsing daily rows.

7. Production Best Practice When computing cumulative metrics across multi-year datasets, always partition by year or quarter (e.g., `Window.partitionBy("year").orderBy("date")`) if the business metric is Year-to-Date (YTD). This resets the running total at year boundaries and allows Spark to distribute the workload across multiple executor nodes in parallel.

8. Real-World Engineering Scenario SaaS recurring revenue dashboards track cumulative Annual Recurring Revenue (ARR) additions over each quarter. A nightly pipeline reads subscription event streams and calculates cumulative new ARR day-by-day to measure sales target pacing.

9. Common Pitfall / Anti-Pattern Using SQL triangular self-joins (e.g. `FROM sales a JOIN sales b ON a.date >= b.date`) instead of a Window function. Triangular joins have $O(N^2)$ complexity, which destroys cluster performance on large datasets, whereas a Window function runs in $O(N \log N)$ time.

10. Tutorial Reference Read more on cumulative window aggregations in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept When a dataset contains multiple historical versions or event updates for an entity and only the latest snapshot is needed, data engineers use the **deduplication pattern**: `Window.partitionBy(entity_id).orderBy(timestamp.desc())` followed by filtering for **`row_number() == 1`**.

2. Detailed Breakdown Execution mechanics of the `row_number() == 1` pattern: 1. **Partitioning:** All records belonging to the same customer ID are routed to the same partition via hash partitioning. 2. **Sorting:** Records within that customer's partition are sorted by timestamp in descending order (newest first). 3. **Numbering:** `row_number()` assigns 1 to the newest record, 2 to the second newest, and so forth. 4. **Filtering:** Applying `.filter(col("rn") == 1)` discards all older historical records, retaining exactly the latest state for every customer.

3. Why the Correct Answer is Correct The requirement states: *"A customer can have multiple records. We only want the newest record."* - `row_number() = 1` partitioned by customer and ordered descending by date reliably extracts the single newest record for each customer. - Therefore, option A is the standard industry pattern.

4. Why Each Distractor is Incorrect - **B (collect()):** Gathers all distributed rows to the driver JVM as a Python list; it performs no filtering or latest-row selection. - **C (count() = 1):** Filtering by `count() = 1` would only retain customers who have *exactly one* record in total, completely discarding customers who have multiple records! - **D (distinct()):** Removes duplicate rows where *all* column values are identical; if customer records have different timestamps or addresses, `distinct()` keeps all of them.

5. Code Example

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

spark = SparkSession.builder.appName("NewestRecordDemo").getOrCreate()

customer_updates = [
    (1, "Alice", "Active", "2023-01-01"),
    (1, "Alice", "Suspended", "2023-02-15"), # Newest for 1
    (2, "Bob", "Active", "2023-01-10")       # Only for 2
]
df = spark.createDataFrame(customer_updates, ["cust_id", "name", "status", "updated_at"])

w = Window.partitionBy("cust_id").orderBy(desc("updated_at"))

latest_df = df.withColumn("rn", row_number().over(w)).filter(col("rn") == 1).drop("rn")
latest_df.show()
# +-------+-----+---------+----------+
# |cust_id| name|   status|updated_at|
# +-------+-----+---------+----------+
# |      1|Alice|Suspended|2023-02-15|
# |      2|  Bob|   Active|2023-01-10|
# +-------+-----+---------+----------+

6. Key Takeaway Filtering for `row_number() = 1` over a partition ordered descending by modification timestamp is the standard, battle-tested pattern to extract the newest record per entity.

7. Production Best Practice If you are working with Delta Lake or Iceberg, you can also perform this deduplication during ingestion using `MERGE INTO` with `WHEN MATCHED THEN UPDATE`. However, when processing raw batch DataFrames, the `row_number() == 1` pattern remains the most universal approach.

8. Real-World Engineering Scenario In CDC (Change Data Capture) pipelines ingesting Debezium Kafka streams from MySQL or Postgres into a data lake, multiple updates arrive for the same entity every hour. A staging job runs `row_number() = 1` to compact the raw CDC stream into a clean snapshot table.

9. Common Pitfall / Anti-Pattern Forgetting to drop the helper column (`.drop("rn")`) before saving the DataFrame to disk. Leaving technical window ranking columns in public analytics tables pollutes the schema.

10. Tutorial Reference Read about deduplication workflows in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept Computing **Top-N items per group** (such as top 3 products in each country) is a classic analytical problem that cannot be solved with a simple `LIMIT 3` because `LIMIT` applies globally to the entire dataset. In distributed SQL and PySpark, this is solved using a **Window specification partitioned by group and ordered descending by metric, evaluated with `rank()` (or `dense_rank()`)**.

2. Detailed Breakdown Implementation architecture: 1. **Window specification:**

   w = Window.partitionBy("country").orderBy(desc("revenue"))

2. Ranking calculation:

   ranked_df = df.withColumn("product_rank", rank().over(w))

3. Filtering Top N:

   top3_df = ranked_df.filter(col("product_rank") <= 3)

• Spark's Catalyst optimizer leverages the WindowExec operator. Within each country partition, products are ordered by revenue, ranked, and filtered down to ranks 1, 2, and 3.

3. Why the Correct Answer is Correct - The requirement has two structural constraints: 1. "Within each country" -> Requires partitioning by country. 2. "Top 3 products by revenue" -> Requires ordering descending by revenue and filtering ranks $\le 3$. - Chaining a **Window specification with `rank()`** is the exact capability designed for this. - Therefore, option B is the correct answer.

4. Why Each Distractor is Incorrect - **A (fillna()):** Imputes missing values; it cannot rank products or select top performers. - **C (dropDuplicates()):** Removes duplicate rows; it cannot calculate top-N metrics. - **D (union()):** Merges DataFrames vertically; it does not rank or filter by revenue.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import rank, desc, col

spark = SparkSession.builder.appName("TopNProducts").getOrCreate()

product_sales = [
    ("USA", "Product_A", 5000),
    ("USA", "Product_B", 3000),
    ("USA", "Product_C", 2000),
    ("USA", "Product_D", 1000), # 4th in USA
    ("UK",  "Product_X", 9000),
    ("UK",  "Product_Y", 8000),
    ("UK",  "Product_Z", 7000),
    ("UK",  "Product_W", 4000)  # 4th in UK
]
df = spark.createDataFrame(product_sales, ["country", "product", "revenue"])

# Partition by country, order by revenue descending
w = Window.partitionBy("country").orderBy(desc("revenue"))

top3_per_country = df.withColumn("rk", rank().over(w)).filter(col("rk") <= 3).drop("rk")
top3_per_country.show()
# +-------+---------+-------+
# |country|  product|revenue|
# +-------+---------+-------+
# |    USA|Product_A|   5000|
# |    USA|Product_B|   3000|
# |    USA|Product_C|   2000|
# |     UK|Product_X|   9000|
# |     UK|Product_Y|   8000|
# |     UK|Product_Z|   7000|
# +-------+---------+-------+

6. Key Takeaway To extract the Top N records within each categorical group, partition by the group, order descending by the metric, and filter for `rank() <= N`.

7. Production Best Practice Clarify with stakeholders how ties should be treated: - If two products tie for 3rd place, should both be included? - Use `rank()` or `dense_rank()`. - Should exactly 3 products be returned under all circumstances? - Use `row_number() <= 3` with a deterministic secondary sort key.

8. Real-World Engineering Scenario In e-commerce merchandising, regional marketing managers feature the top 5 trending products per country on localized homepage banners. A scheduled PySpark job runs nightly to compute top products by sales volume using `Window + rank()` and pushes the results to Redis caches.

9. Common Pitfall / Anti-Pattern Attempting to loop through unique country values in a Python `for` loop, filtering the DataFrame for each country, sorting, applying `.limit(3)`, and unioning them back together. This launches dozens of separate Spark jobs and destroys cluster parallelism, whereas a single `Window + rank()` achieves the result in a single distributed pass.

10. Tutorial Reference Read about Top-N group ranking in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In the PySpark Window API, **`Window.partitionBy("column")`** defines the boundary or slice across which window calculations (such as ranking, running totals, lead, and lag) are evaluated. When applied to an entity like `department`, it indicates that the calculation is performed independently within each departmental group.

2. Detailed Breakdown Analyzing the code snippet:

Window.partitionBy("department")

• This is part of pyspark.sql.window.Window.
• It dictates that rows with the same department value are grouped together for window calculation purposes.
• Any subsequent ranking function (such as rank(), dense_rank(), or row_number()) chained with an orderBy will calculate ranks *within each department*, resetting the rank back to 1 for each new department.
• It does NOT repartition the physical files on disk or change the DataFrame partition count; it configures the window frame boundary.

3. Why the Correct Answer is Correct - `Window.partitionBy("department")` establishes group boundaries for window operations. - Combined with an `orderBy`, it is the foundational step for **Ranking within groups** (e.g., ranking employees by salary within their department). - Therefore, option C is the correct business requirement.

4. Why Each Distractor is Incorrect - **A (Reading CSV files):** Reading CSV files uses `spark.read.csv()`; window objects play no role in file ingestion. - **B (Broadcasting data):** Broadcasting uses `broadcast(df)` to replicate small lookup tables across executors; window partitioning is completely unrelated. - **D (Creating partitions in storage):** Partitioning storage files is achieved using `df.write.partitionBy("department")`, not `Window.partitionBy()`.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import rank, desc

spark = SparkSession.builder.appName("WindowPartitionDemo").getOrCreate()

employees = [
    ("Alice", "Engineering", 120000),
    ("Bob", "Engineering", 110000),
    ("Charlie", "Sales", 95000),
    ("Diana", "Sales", 90000)
]
df = spark.createDataFrame(employees, ["name", "department", "salary"])

# Window partitioned by department for ranking within each group
w = Window.partitionBy("department").orderBy(desc("salary"))

ranked_df = df.withColumn("dept_rank", rank().over(w))
ranked_df.show()
# +-------+-----------+------+---------+
# |   name| department|salary|dept_rank|
# +-------+-----------+------+---------+
# |  Alice|Engineering|120000|        1|
# |    Bob|Engineering|110000|        2|
# |Charlie|      Sales| 95000|        1|
# |  Diana|      Sales| 90000|        2|
# +-------+-----------+------+---------+
# Notice dept_rank resets to 1 when department changes!

6. Key Takeaway `Window.partitionBy()` defines the grouping boundaries for windowed computations, ensuring ranking and aggregation functions reset for each unique partition key.

7. Production Best Practice Distinguish clearly between: - `Window.partitionBy()`: Defines logical windows for analytical calculations. - `df.repartition()`: Changes the in-memory distribution of Spark partitions across cluster nodes. - `df.write.partitionBy()`: Creates physical subdirectories on disk (e.g. `department=Engineering/`). Confusing these three distinct concepts is a major source of confusion for junior engineers.

8. Real-World Engineering Scenario In university administration systems, student GPAs must be ranked within their specific major or academic department to determine class valedictorians and dean's list honors. The pipeline partitions by `academic_department` to evaluate ranks within each cohort.

9. Common Pitfall / Anti-Pattern Assuming `Window.partitionBy()` controls how data is saved to S3 or ADLS. Writing `df.withColumn("rn", row_number().over(Window.partitionBy("dept"))).write.parquet("...")` does NOT save partitioned folders on disk unless `.partitionBy("dept")` is called on the `DataFrameWriter`.

10. Tutorial Reference Read about window partitioning semantics in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In sequential data analysis, comparing a record's current value with its immediate chronological predecessor requires retrieving data from the prior row. In PySpark, this is achieved using the **`lag()`** window function.

2. Detailed Breakdown In PySpark: - Syntax:

  w = Window.partitionBy("route_id").orderBy("shipment_date")
  df.withColumn("prev_cost", lag("cost", 1).over(w)) \
    .withColumn("cost_variance", col("cost") - col("prev_cost"))

• Mechanism: - Spark sorts the records by shipment_date within each route_id partition. - For each shipment, lag("cost", 1) looks back by 1 position and projects the previous shipment's cost into the current row. - For the first shipment on each route, prev_cost evaluates to NULL.

3. Why the Correct Answer is Correct The requirement states: *"compare current shipment cost with previous shipment cost"*. - "Previous shipment cost" is located in the preceding row. - `lag()` retrieves values from preceding rows. - Therefore, option D is the most suitable function.

4. Why Each Distractor is Incorrect - **A (lead()):** Looks forward to the *next* shipment cost, not the previous one. - **B (max()):** Finds the maximum cost across all shipments; it does not isolate the immediate previous shipment. - **C (count()):** Tallies total shipments; it cannot access cost values from prior rows.

5. Code Example

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

spark = SparkSession.builder.appName("ShipmentCostLagDemo").getOrCreate()

shipments = [
    ("Route_100", "2023-04-01", 1200.0),
    ("Route_100", "2023-04-10", 1350.0),
    ("Route_100", "2023-04-20", 1100.0)
]
df = spark.createDataFrame(shipments, ["route_id", "shipment_date", "cost"])

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

result = df.withColumn("prev_cost", lag("cost", 1).over(w)) \
           .withColumn("cost_diff", col("cost") - col("prev_cost"))
result.show()
# +---------+-------------+------+---------+---------+
# | route_id|shipment_date|  cost|prev_cost|cost_diff|
# +---------+-------------+------+---------+---------+
# |Route_100|   2023-04-01|1200.0|     NULL|     NULL|
# |Route_100|   2023-04-10|1350.0|   1200.0|    150.0|
# |Route_100|   2023-04-20|1100.0|   1350.0|   -250.0|
# +---------+-------------+------+---------+---------+

6. Key Takeaway Use `lag()` to access prior records in an ordered sequence to compute variance, deltas, and percentage changes against previous events.

7. Production Best Practice Provide a meaningful default value when handling the initial row's null return if needed by downstream math: `lag("cost", 1, 0.0).over(w)`. However, for cost variance or percentage change calculations, leaving the first row as `NULL` is generally preferred because an initial event has no genuine baseline for variance.

8. Real-World Engineering Scenario In freight rate auditing, logistics companies monitor fuel price volatility. When carrier invoices arrive, an automated Spark job flags any shipment whose cost exceeds the prior shipment on that lane by more than 15% for manual audit.

9. Common Pitfall / Anti-Pattern Sorting by a non-unique column without a tie-breaker. If two shipments on the same route share the same date, Spark cannot guarantee which shipment came first unless an incremental shipment ID is included in `orderBy`.

10. Tutorial Reference Read about comparing sequential records in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In predictive maintenance and scheduling systems, viewing upcoming scheduled events alongside historical maintenance records requires looking forward in the timeline. In PySpark, looking ahead to the next sequential row in an ordered window is accomplished using **`lead()`**.

2. Detailed Breakdown In PySpark: - Syntax:

  w = Window.partitionBy("machine_id").orderBy("maintenance_date")
  df.withColumn("next_maintenance", lead("maintenance_date", 1).over(w))

• Under the hood: - Spark evaluates rows per machine_id. - For each maintenance record, lead("maintenance_date", 1) inspects the subsequent row in the partition and extracts its date. - For the latest maintenance event on a machine, lead() returns NULL, indicating that no future maintenance has been scheduled yet.

3. Why the Correct Answer is Correct The requirement states: *"the next machine maintenance date for every machine"*. - "Next" means looking forward in time to the succeeding row. - `lead()` fetches values from subsequent rows within an ordered partition. - Therefore, option A is the correct function.

4. Why Each Distractor is Incorrect - **B (sum()):** Sums numerical values; it cannot manipulate dates or access future rows. - **C (avg()):** Calculates arithmetic means; it cannot retrieve future maintenance dates. - **D (lag()):** Looks backward to the *previous* maintenance date, which is the exact opposite of "next".

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import lead

spark = SparkSession.builder.appName("MaintenanceLeadDemo").getOrCreate()

maintenance = [
    ("M_01", "2023-01-10"),
    ("M_01", "2023-04-15"),
    ("M_01", "2023-08-20"),
    ("M_02", "2023-02-05")
]
df = spark.createDataFrame(maintenance, ["machine_id", "maintenance_date"])

w = Window.partitionBy("machine_id").orderBy("maintenance_date")

result = df.withColumn("next_maintenance_date", lead("maintenance_date", 1).over(w))
result.show()
# +----------+----------------+---------------------+
# |machine_id|maintenance_date|next_maintenance_date|
# +----------+----------------+---------------------+
# |      M_01|      2023-01-10|           2023-04-15|
# |      M_01|      2023-04-15|           2023-08-20|
# |      M_01|      2023-08-20|                 NULL|
# |      M_02|      2023-02-05|                 NULL|
# +----------+----------------+---------------------+

6. Key Takeaway The `lead()` window function retrieves upcoming values from subsequent rows in an ordered partition, ideal for scheduling and tracking future events.

7. Production Best Practice When computing the number of days until the next maintenance event, combine `lead()` with `datediff()`: `datediff(lead("maintenance_date", 1).over(w), col("maintenance_date"))`.

8. Real-World Engineering Scenario In aviation fleet maintenance, commercial airlines monitor turbine inspection intervals. Aviation safety regulations mandate that aircraft engines undergo overhaul every $N$ operating cycles. ETL pipelines use `lead()` to calculate projected cycles until the next hangar overhaul.

9. Common Pitfall / Anti-Pattern Assuming `lead()` without an `orderBy` clause will work. Window analytical offset functions (`lead`, `lag`) strictly require an ordered sequence; omitting `orderBy` throws an `AnalysisException`.

10. Tutorial Reference Explore offset window functions in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept Extracting the first (earliest) historical event per entity is the inverse of the latest-record pattern. In PySpark, this is executed by partitioning by entity, ordering **ascending** by timestamp, and filtering for **`row_number() == 1`**.

2. Detailed Breakdown In PySpark: - Syntax:

  w = Window.partitionBy("customer_id").orderBy(col("txn_timestamp").asc())
  first_txn = df.withColumn("rn", row_number().over(w)).filter(col("rn") == 1)

• Execution flow: - Spark routes customer records to partitions by hashing customer_id. - Inside each partition, transactions are sorted chronologically from oldest to newest (ASC). - row_number() assigns 1 to the very first transaction recorded for that customer. - Filtering for rn = 1 extracts the complete initial transaction record for every customer across the entire dataset.

3. Why the Correct Answer is Correct - The requirement is to identify the first transaction (the complete row) for every customer. - `row_number()` over an ascending chronological window identifies row 1 as the earliest transaction. - Therefore, option B is the standard and commonly used function.

4. Why Each Distractor is Incorrect - **A (distinct()):** Removes duplicate rows across columns; it does not sort by timestamp or isolate the initial transaction. - **C (collect()):** Gathers distributed rows into the driver memory as a Python list, which can crash the driver and does not filter by time. - **D (union()):** Combines two DataFrames vertically; it has no ranking or filtering capability.

5. Code Example

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

spark = SparkSession.builder.appName("FirstTxnDemo").getOrCreate()

txns = [
    ("Cust_A", "TXN_01", "2023-01-05", 50.0), # First transaction
    ("Cust_A", "TXN_02", "2023-02-14", 120.0),
    ("Cust_B", "TXN_03", "2023-01-10", 300.0)  # First transaction
]
df = spark.createDataFrame(txns, ["customer_id", "txn_id", "txn_date", "amount"])

# Window ordered ASCENDING by date
w = Window.partitionBy("customer_id").orderBy(asc("txn_date"))

first_txns = df.withColumn("rn", row_number().over(w)).filter(col("rn") == 1).drop("rn")
first_txns.show()
# +-----------+------+----------+------+
# |customer_id|txn_id|  txn_date|amount|
# +-----------+------+----------+------+
# |     Cust_A|TXN_01|2023-01-05|  50.0|
# |     Cust_B|TXN_03|2023-01-10| 300.0|
# +-----------+------+----------+------+

6. Key Takeaway To extract the initial historical record for each entity, partition by entity, order ascending by timestamp, and filter for `row_number() == 1`.

7. Production Best Practice When extracting customer onboarding or first-order attribution data in marketing pipelines, save the resulting `first_transactions` table as a permanent lookup dimension. Joining new incoming transactions to this static lookup avoids recomputing historical window partitions over billions of legacy transactions on every daily run.

8. Real-World Engineering Scenario In customer acquisition cost (CAC) and marketing attribution analytics, growth teams need to identify the exact marketing campaign or promo code used on a customer's very first order. Data engineers run `row_number() == 1` ordered ascending by order timestamp to capture the acquisition source.

9. Common Pitfall / Anti-Pattern Using `min("txn_date")` inside `groupBy("customer_id")` and attempting to select non-aggregated attributes (e.g. `promo_code`). Standard SQL and Spark SQL do not allow selecting unaggregated columns alongside `groupBy()`. Using `row_number()` preserves all columns seamlessly.

10. Tutorial Reference Read about first/last record extraction in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept Calculating sequential change, growth rate, or period-over-period variance (such as day-over-day sales change) requires accessing the measurement from the **prior time period**. In PySpark, this historical lookup is executed using **`lag()`**.

2. Detailed Breakdown To compute daily sales change: - Formula: $$\text{Daily Change} = \text{Current Day Sales} - \text{Previous Day Sales}$$ - PySpark Implementation:

  w = Window.partitionBy("store_id").orderBy("sale_date")
  df.withColumn("prev_day_sales", lag("sales_amount", 1).over(w)) \
    .withColumn("sales_change", col("sales_amount") - col("prev_day_sales"))

• Execution: - Spark buffers the ordered daily records. - For each day, lag("sales_amount", 1) extracts the sales figure from the preceding day. - A simple column subtraction calculates the daily variance.

3. Why the Correct Answer is Correct The requirement asks: *"how much sales changed compared to the previous day"*. - Retrieving the sales from the previous day requires the **`lag()`** function. - Subtracting the lagged value from current sales produces the daily change. - Therefore, option C is the correct function.

4. Why Each Distractor is Incorrect - **A (cache()):** Persists a DataFrame in executor memory; it does not calculate metrics or access previous rows. - **B (countDistinct()):** Counts unique values; it does not compute daily variances. - **D (dense_rank()):** Calculates ordinal ranking positions; it cannot retrieve previous day sales figures.

5. Code Example

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

spark = SparkSession.builder.appName("DailySalesChangeDemo").getOrCreate()

daily_sales = [
    ("Store_1", "2023-06-01", 5000.0),
    ("Store_1", "2023-06-02", 6200.0), # +1200
    ("Store_1", "2023-06-03", 5800.0)  # -400
]
df = spark.createDataFrame(daily_sales, ["store_id", "sale_date", "sales_amount"])

w = Window.partitionBy("store_id").orderBy("sale_date")

result = df.withColumn("prev_sales", lag("sales_amount", 1).over(w)) \
           .withColumn("sales_change", col("sales_amount") - col("prev_sales"))
result.show()
# +--------+----------+------------+----------+------------+
# |store_id| sale_date|sales_amount|prev_sales|sales_change|
# +--------+----------+------------+----------+------------+
# | Store_1|2023-06-01|      5000.0|      NULL|        NULL|
# | Store_1|2023-06-02|      6200.0|    5000.0|      1200.0|
# | Store_1|2023-06-03|      5800.0|    6200.0|      -400.0|
# +--------+----------+------------+----------+------------+

6. Key Takeaway The `lag()` window function retrieves values from preceding time periods, enabling period-over-period variance and delta calculations.

7. Production Best Practice To compute percentage change rather than absolute dollar change:

percentage_change = (col("sales_amount") - col("prev_sales")) / col("prev_sales") * 100

Always guard against division by zero using when(col("prev_sales") != 0, percentage_change).otherwise(None).

8. Real-World Engineering Scenario In retail demand forecasting, supply chain systems track sudden spikes or drops in daily product velocity. If sales drop by more than 50% compared to the previous day, the pipeline flags potential stockout anomalies or register outages.

9. Common Pitfall / Anti-Pattern Assuming daily records are contiguous without checking for calendar gaps. If a store is closed on Sundays and has no record, `lag(1)` on Monday compares Monday against Saturday. If strict 1-day comparison is required, ensure missing dates are imputed with zero sales beforehand.

10. Tutorial Reference Learn about time-series comparison patterns in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept When ordering entities where ties receive the same rank and subsequent ranks proceed **strictly consecutively without numerical gaps** (e.g. 1, 2, 2, 3), the required function is **`dense_rank()`**.

2. Detailed Breakdown Comparing the ranking functions: - **`dense_rank()`**: - Scores: `[98, 95, 95, 90]` -> Ranks: `[1, 2, 2, 3]`. - Ranks advance consecutively without gaps. - **`rank()`**: - Scores: `[98, 95, 95, 90]` -> Ranks: `[1, 2, 2, 4]`. - Leaves a gap (rank 3 is skipped). - **`row_number()`**: - Scores: `[98, 95, 95, 90]` -> Ranks: `[1, 2, 3, 4]`. - Breaks ties arbitrarily, never assigning duplicate ranks.

3. Why the Correct Answer is Correct The requirement explicitly states: 1. *"Employees with equal scores should receive consecutive rankings."* 2. *"without gaps."* This exact behavior defines **`dense_rank()`**. Therefore, option D is the correct function.

4. Why Each Distractor is Incorrect - **A (rank()):** Leaves gaps whenever ties occur (e.g. 1, 2, 2, 4), directly violating "without gaps". - **B (row_number()):** Assigns unique sequential numbers without recognizing ties, giving tied employees different ranks. - **C (lead()):** Looks ahead to future rows; it does not compute performance ranks.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import dense_rank, desc

spark = SparkSession.builder.appName("EmployeeDenseRankDemo").getOrCreate()

performance = [
    ("Alice", 98),
    ("Bob", 95),
    ("Charlie", 95), # Tied with Bob
    ("David", 90)
]
df = spark.createDataFrame(performance, ["employee_name", "score"])

w = Window.orderBy(desc("score"))

result = df.withColumn("dense_rank", dense_rank().over(w))
result.show()
# +-------------+-----+----------+
# |employee_name|score|dense_rank|
# +-------------+-----+----------+
# |        Alice|   98|         1|
# |          Bob|   95|         2|
# |      Charlie|   95|         2|
# |        David|   90|         3|
# +-------------+-----+----------+
# Notice David receives rank 3 with no gap!

6. Key Takeaway `dense_rank()` assigns tied values the same ranking position and advances subsequent ranks consecutively without numerical gaps.

7. Production Best Practice When designing compensation and bonus tiers, `dense_rank()` ensures that everyone achieving the $N$-th unique performance tier is rewarded fairly without skipping bonus tiers due to peer ties.

8. Real-World Engineering Scenario In corporate HR annual reviews, compensation bands are tied to performance quartiles. Employees are ranked by review score using `dense_rank()` so that tied high-performers share the top tier while subsequent performers fall into tier 2 rather than being penalized by skipped numbers.

9. Common Pitfall / Anti-Pattern Assuming `dense_rank()` limits the number of output rows. If you filter for `dense_rank() <= 10`, you are filtering for the top 10 *distinct score values*, which could return 50 employees if multiple employees share scores. If an exact count of 10 employees is mandated, use `row_number() <= 10`.

10. Tutorial Reference Review ranking functions in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In event streaming, fraud detection, and transactional auditing, detecting anomalies often requires comparing an incoming event against its immediate predecessor (e.g., sudden amount jump or geographical impossibility). In PySpark, appending the previous transaction's attributes to the current row is performed using **`lag()`**.

2. Detailed Breakdown In PySpark: - Syntax:

  w = Window.partitionBy("card_id").orderBy("txn_timestamp")
  df.withColumn("prev_txn_amount", lag("amount", 1).over(w))

• Execution: - Spark partitions transactions by credit card or account identifier. - Transactions are sorted chronologically. - For each transaction, lag("amount", 1) inspects the preceding transaction on that card and appends its dollar amount. - This allows downstream feature engineering models to calculate features like current_amount / prev_txn_amount or time difference between transactions.

3. Why the Correct Answer is Correct The requirement states: *"the previous transaction amount available in every record"*. - "Previous transaction amount" refers to the dollar amount in the immediately preceding row. - `lag("amount", 1)` projects that preceding value into the current row. - Therefore, option A is the appropriate function.

4. Why Each Distractor is Incorrect - **B (lead()):** Retrieves the *subsequent* (future) transaction amount, not the previous one. - **C (row_number()):** Generates an integer row counter (1, 2, 3); it cannot extract currency amounts from prior transactions. - **D (distinct()):** Removes duplicate rows; it performs no inter-row lookups.

5. Code Example

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

spark = SparkSession.builder.appName("FraudLagDemo").getOrCreate()

txns = [
    ("Card_999", "2023-12-01 10:00:00", 15.0),
    ("Card_999", "2023-12-01 10:02:00", 2500.0), # Rapid surge
    ("Card_999", "2023-12-01 11:30:00", 30.0)
]
df = spark.createDataFrame(txns, ["card_id", "txn_time", "amount"])

w = Window.partitionBy("card_id").orderBy("txn_time")

result = df.withColumn("prev_amount", lag("amount", 1).over(w))
result.show()
# +--------+-------------------+------+-----------+
# | card_id|           txn_time|amount|prev_amount|
# +--------+-------------------+------+-----------+
# |Card_999|2023-12-01 10:00:00|  15.0|       NULL|
# |Card_999|2023-12-01 10:02:00|2500.0|       15.0|
# |Card_999|2023-12-01 11:30:00|  30.0|     2500.0|
# +--------+-------------------+------+-----------+

6. Key Takeaway Use `lag()` to project attributes from previous events onto the current event record, enabling inter-event comparison and anomaly detection.

7. Production Best Practice In machine learning feature pipelines for fraud detection, lag operations should also capture the timestamp difference:

unix_timestamp(col("txn_time")) - unix_timestamp(lag("txn_time", 1).over(w))

This produces the "seconds since last transaction" feature, which is a powerful signal for card-not-present fraud detection.

8. Real-World Engineering Scenario Credit card fraud models look for "card testing" behavior—where a fraudster tests a stolen card with a $1.00 charge followed seconds later by a $2,500.00 luxury purchase. PySpark ETL pipelines compute `lag("amount")` and time delta to flag suspicious transaction velocity spikes in real time.

9. Common Pitfall / Anti-Pattern Failing to partition by card ID or user ID. If `partitionBy` is omitted, transactions from different customers interleave chronologically, causing Card B's transaction to lag against Card A's transaction!

10. Tutorial Reference Master sequential feature engineering in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In subscription analytics, customer success monitoring, and contract management, projecting an upcoming contractual milestone onto the active subscription row requires looking forward in the chronological sequence. In PySpark, this is achieved using **`lead()`**.

2. Detailed Breakdown In PySpark: - Syntax:

  w = Window.partitionBy("customer_id").orderBy("renewal_date")
  df.withColumn("next_renewal_date", lead("renewal_date", 1).over(w))

• Execution: - Spark orders all renewal events per customer chronologically. - For each active contract row, lead("renewal_date", 1) peeks ahead to the subsequent renewal event and extracts its scheduled date. - If no future renewal exists in the dataset, lead() returns NULL.

3. Why the Correct Answer is Correct The requirement explicitly states: *"the next upcoming customer renewal date visible in the current row"*. - "Next upcoming" refers to the succeeding row in chronological order. - `lead()` retrieves values from subsequent rows within an ordered window. - Therefore, option B is the correct function to apply.

4. Why Each Distractor is Incorrect - **A (lag()):** Looks backward to past renewal dates, not forward to upcoming dates. - **C (rank()):** Generates an ordinal number (1, 2, 3); it cannot retrieve date values. - **D (sum()):** Computes numerical totals; it does not access subsequent rows.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import lead

spark = SparkSession.builder.appName("RenewalLeadDemo").getOrCreate()

renewals = [
    ("Cust_A", "2023-01-01"),
    ("Cust_A", "2024-01-01"),
    ("Cust_A", "2025-01-01"),
    ("Cust_B", "2023-06-01")
]
df = spark.createDataFrame(renewals, ["customer_id", "renewal_date"])

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

result = df.withColumn("next_renewal_date", lead("renewal_date", 1).over(w))
result.show()
# +-----------+------------+-----------------+
# |customer_id|renewal_date|next_renewal_date|
# +-----------+------------+-----------------+
# |     Cust_A|  2023-01-01|       2024-01-01|
# |     Cust_A|  2024-01-01|       2025-01-01|
# |     Cust_A|  2025-01-01|             NULL|
# |     Cust_B|  2023-06-01|             NULL|
# +-----------+------------+-----------------+

6. Key Takeaway The `lead()` window function retrieves upcoming values from subsequent records, enabling pipelines to project future renewal dates and upcoming contract milestones.

7. Production Best Practice When feeding customer success dashboards (e.g. Salesforce or HubSpot), calculate the days remaining until the upcoming renewal: `datediff(col("next_renewal_date"), current_date())`. This allows account executives to trigger automated renewal reminders 90, 60, and 30 days in advance.

8. Real-World Engineering Scenario In enterprise SaaS billing platforms, customer contracts renew annually. Data pipelines evaluate current active subscriptions and project the `next_renewal_date` to build executive ARR renewal forecast models for financial reporting.

9. Common Pitfall / Anti-Pattern Assuming `lead()` and `lag()` can be executed without an `orderBy()` specification. Window offset functions require a strictly defined sequence order; omitting `orderBy` causes Spark to throw an `AnalysisException`.

10. Tutorial Reference Read about window offset functions in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In analytics reporting, identifying the top performers (Top N) across an entire dataset or within specific sub-groups is achieved using a **ranking function (`rank()` or `dense_rank()`) over an ordered window, followed by a filter predicate (`rank <= N`)**.

2. Detailed Breakdown To extract the Top 5 customers by yearly spending: 1. **Aggregate spending:** First aggregate transaction logs into annual customer totals:

   annual_spending = df.groupBy("customer_id").agg(sum("amount").alias("yearly_spending"))

2. Apply ranking window:

   w = Window.orderBy(desc("yearly_spending"))
   ranked_df = annual_spending.withColumn("spending_rank", rank().over(w))

3. Filter Top 5:

   top5 = ranked_df.filter(col("spending_rank") <= 5)

• This pattern guarantees that ties in the 5th position are handled according to business ranking rules.

3. Why the Correct Answer is Correct The prompt asks for the most common window-based approach to get the Top 5 customers by yearly spending: - The standard window-based pattern is computing ranks with **`rank()` or `dense_rank()`** and applying a **`filter(rank <= 5)`**. - Therefore, option C is the correct approach.

4. Why Each Distractor is Incorrect - **A (union()):** Combines two DataFrames vertically; it does not rank or filter records. - **B (collect()):** Pulls distributed records into the driver memory; it does not perform ranking or top-N selection. - **D (fillna()):** Imputes missing values; it has no ranking capability.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import rank, desc, col

spark = SparkSession.builder.appName("Top5CustomersDemo").getOrCreate()

customers = [
    ("Cust_1", 12500.0),
    ("Cust_2", 8900.0),
    ("Cust_3", 15400.0),
    ("Cust_4", 21000.0),
    ("Cust_5", 9400.0),
    ("Cust_6", 4500.0)
]
df = spark.createDataFrame(customers, ["customer_id", "yearly_spending"])

# Rank customers globally by spending descending
w = Window.orderBy(desc("yearly_spending"))

top5 = df.withColumn("rank", rank().over(w)).filter(col("rank") <= 5)
top5.show()
# +-----------+---------------+----+
# |customer_id|yearly_spending|rank|
# +-----------+---------------+----+
# |     Cust_4|        21000.0|   1|
# |     Cust_3|        15400.0|   2|
# |     Cust_1|        12500.0|   3|
# |     Cust_5|         9400.0|   4|
# |     Cust_2|         8900.0|   5|
# +-----------+---------------+----+

6. Key Takeaway The standard window-based Top-N pattern combines an ordered ranking function (`rank()` or `dense_rank()`) with a filter condition (`rank <= N`).

7. Production Best Practice If you need global Top 5 without partitioning by group and don't require handling ties with specific window semantics, `df.orderBy(desc("yearly_spending")).limit(5)` is physically more efficient because Spark optimizes it with a local top-N heap (`TakeOrderedAndProjectExec`). However, when computing Top 5 within partitions (e.g. Top 5 customers per country), the `rank() + filter` window pattern is strictly required.

8. Real-World Engineering Scenario In VIP customer loyalty programs (e.g. airline elite status or private banking tiers), marketing teams identify the top-spending customers each quarter to send personalized luxury gifts and invitations to executive summits.

9. Common Pitfall / Anti-Pattern Using `row_number()` when the business requires all customers tied for 5th place to be recognized. `row_number()` will arbitrarily cut off ties, leaving out equally qualified customers. Use `rank()` or `dense_rank()` when tie-inclusion is required.

10. Tutorial Reference Read more about Top-N ranking in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).

1. Core Concept In relational algebra and distributed data engineering, **Window Functions** represent one of the most powerful and frequently used analytical tools. Unlike standard `groupBy()` aggregations—which collapse multiple rows into a single summary row per group—window functions evaluate calculations across partitioned slices of data **while preserving every individual row in the dataset**.

2. Detailed Breakdown Why Window Functions are central to real-world data engineering: 1. **Latest / Deduplication:** `row_number() = 1` over timestamp desc cleanly extracts current state in CDC and SCD pipelines. 2. **Ranking:** `rank()` and `dense_rank()` solve Top-N leaderboards within groups. 3. **Period-over-Period Comparisons:** `lag()` and `lead()` enable day-over-day, month-over-month variance without self-joins. 4. **Cumulative Calculations:** `sum() over w` calculates running balances, moving averages, and cumulative totals. 5. **Preserving Granularity:** Business reports frequently require displaying granular transaction attributes alongside group benchmarks (e.g., individual employee salary alongside departmental average salary). A window function like `avg("salary").over(Window.partitionBy("dept"))` appends this benchmark seamlessly.

3. Why the Correct Answer is Correct The interviewer's statement is completely accurate: - Latest records, ranking, trends, comparisons, and cumulative sums are precisely the core use cases of window functions. - Window functions perform these calculations across partitioned slices while preserving the full row-level granularity of the dataset. - Therefore, option D is the correct and comprehensive technical evaluation.

4. Why Each Distractor is Incorrect - **A (Incorrect - Window functions can only be used on streaming):** Window functions are fully supported in batch DataFrames as well as Structured Streaming (with watermarks). They are not limited to streaming. - **B (Incorrect - GroupBy with aggregations always replaces):** `groupBy()` collapses rows and loses individual record details. It cannot directly compute running totals, row-level lead/lag, or Top-N per group without complex, expensive self-joins. - **C (Incorrect - Window functions cannot compute running totals or ranking):** Running totals and ranking are the primary canonical capabilities of window functions.

5. Code Example

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import avg, col, round as spark_round

spark = SparkSession.builder.appName("WindowGranularityDemo").getOrCreate()

employees = [
    ("Alice", "Engineering", 120000),
    ("Bob", "Engineering", 100000),
    ("Charlie", "Sales", 90000)
]
df = spark.createDataFrame(employees, ["name", "dept", "salary"])

# Window function calculates departmental average while keeping every employee row
w = Window.partitionBy("dept")

result = df.withColumn("dept_avg_salary", spark_round(avg("salary").over(w), 2)) \
           .withColumn("diff_from_avg", col("salary") - col("dept_avg_salary"))
result.show()
# +-------+-----------+------+---------------+-------------+
# |   name|       dept|salary|dept_avg_salary|diff_from_avg|
# +-------+-----------+------+---------------+-------------+
# |  Alice|Engineering|120000|       110000.0|      10000.0|
# |    Bob|Engineering|100000|       110000.0|     -10000.0|
# |Charlie|      Sales| 90000|        90000.0|          0.0|
# +-------+-----------+------+---------------+-------------+

6. Key Takeaway Window functions perform advanced analytical calculations across defined data partitions without collapsing the underlying dataset's row granularity.

7. Production Best Practice Always be mindful of memory consumption when using window functions. An unpartitioned window (`Window.orderBy(...)`) forces the entire distributed dataset onto a single executor core, which can cause severe bottlenecking or Out-Of-Memory errors. Ensure a high-cardinality `partitionBy` column is included whenever possible.

8. Real-World Engineering Scenario In enterprise data warehousing, over 70% of analytical transformation SQL queries for financial reconciliation, fraud detection, customer churn, and executive BI leaderboards rely heavily on window functions rather than simple group-by reductions.

9. Common Pitfall / Anti-Pattern Relying on self-joins to achieve what a window function can do natively. Joining a table to itself on `t1.group = t2.group AND t1.time >= t2.time` creates quadratic $O(N^2)$ data explosion, whereas a window function executes in linear/log-linear time.

10. Tutorial Reference Master all aspects of window functions in [PySpark DataFrame Window Functions](/tutorials/pyspark/df-window-functions).