Top Data engineer interview questions

The data is small on purpose, and each table carries the edge case its questions are about: tied salaries, a missing month, two transactions at the same timestamp, a customer with no orders in February.

-- Sample data shared by every SQL answer in the post.
DROP TABLE IF EXISTS emp, logins, orders, txns, usage_monthly, accounts,
  account_audit, clicks, monthly_revenue, users, events, customer_raw,
  shipments, order_header, daily_sales, contacts, app_events, sellers,
  product_sales, visits, app_users, person, ride_users, trips, parcels,
  web_log, subscriptions, tasks, scores CASCADE;

CREATE TABLE emp (
  emp_id     int PRIMARY KEY,
  name       text NOT NULL,
  dept       text NOT NULL,
  salary     numeric(10,2) NOT NULL,
  manager_id int REFERENCES emp(emp_id),
  hire_date  date NOT NULL
);
INSERT INTO emp VALUES
  (1, 'Ranga',    'ENG',   250000, NULL, '2015-01-10'),
  (2, 'Nishanth', 'ENG',   180000, 1,    '2017-03-01'),
  (3, 'Meena',    'ENG',   180000, 1,    '2016-07-15'),
  (4, 'Vinod',    'ENG',   150000, 2,    '2019-11-20'),
  (5, 'Vasundra', 'SALES', 120000, 1,    '2018-02-01'),
  (6, 'Raja',     'SALES', 120000, 5,    '2020-05-05'),
  (7, 'Lakshmi',  'SALES', 95000,  5,    '2021-09-09'),
  (8, 'Kavi',     'HR',    90000,  1,    '2019-01-01'),
  (9, 'Manoj',    'ENG',   190000, 2,    '2022-04-04');   -- earns more than manager Nishanth

CREATE TABLE logins (user_id int, login_ts timestamp);
INSERT INTO logins VALUES
  (1,'2026-10-01 09:00'),(1,'2026-10-03 10:00'),(1,'2026-10-05 08:00'),
  (2,'2026-10-02 11:00'),
  (3,'2026-10-01 07:00'),(3,'2026-10-04 07:30');

CREATE TABLE orders (order_id int PRIMARY KEY, customer_id int, order_date date, amount numeric(10,2));
INSERT INTO orders VALUES
  (101, 1, '2026-01-05', 100), (102, 1, '2026-01-06', 50), (103, 1, '2026-01-07', 70),
  (104, 1, '2026-02-10', 40),
  (105, 2, '2026-01-15', 300),
  (106, 3, '2026-02-01', 80),  (107, 3, '2026-02-02', 20),
  (108, 4, '2026-01-20', 60),
  (109, 1, '2026-01-07', 10),  -- second order on the same day as 103
  (110, 1, '2025-06-10', 200), (111, 1, '2025-08-10', 150), (112, 1, '2025-10-10', 100),
  (113, 2, '2025-07-01', 50);   -- 2025 history for the baseline in question 14

CREATE TABLE txns (txn_id int, account_id int, txn_ts timestamp, amount numeric(10,2));
INSERT INTO txns VALUES                  -- txn_id records arrival order
  (1, 10,'2026-10-01 09:00:00', 500), (2, 10,'2026-10-01 09:00:00', -200),
  (3, 10,'2026-10-02 10:00:00', -50), (4, 20,'2026-10-01 12:00:00', 1000);

CREATE TABLE usage_monthly (user_id int, month date, usage int);
INSERT INTO usage_monthly VALUES
  (1,'2026-01-01',10),(1,'2026-02-01',12),(1,'2026-03-01',15),(1,'2026-04-01',9),
  (2,'2026-01-01',30),(2,'2026-02-01',25),(2,'2026-03-01',40),(2,'2026-04-01',45);

CREATE TABLE accounts (account_id int PRIMARY KEY, holder text);
INSERT INTO accounts VALUES (1,'Lakshmi'),(2,'Kavi'),(3,'Yashu'),(4,'Anada');
CREATE TABLE account_audit (account_id int, status text, changed_at timestamp);
INSERT INTO account_audit VALUES
  (1,'ACTIVE','2026-01-01'),(1,'SUSPENDED','2026-03-01'),
  (2,'SUSPENDED','2026-01-01'),(2,'ACTIVE','2026-02-01'),
  (3,'CLOSED','2026-02-15');            -- account 4 has no audit rows

CREATE TABLE clicks (user_id int, event_ts timestamp);
INSERT INTO clicks VALUES
  (1,'2026-10-01 09:00'),(1,'2026-10-01 09:10'),(1,'2026-10-01 09:55'),
  (1,'2026-10-01 10:05'),(2,'2026-10-01 12:00'),(2,'2026-10-01 12:20'),
  (2,'2026-10-01 12:50');                -- exactly 30 minutes later: same session

CREATE TABLE monthly_revenue (month date, product_id int, revenue numeric(10,2));
INSERT INTO monthly_revenue VALUES
  ('2026-01-01',1,100),('2026-02-01',1,0),('2026-04-01',1,150),   -- no March row
  ('2026-01-01',2,40),('2026-02-01',2,60),('2026-03-01',2,90);

CREATE TABLE users (user_id int PRIMARY KEY, signup_ts timestamp);
INSERT INTO users VALUES (1,'2026-01-03'),(2,'2026-01-20'),(3,'2026-02-02'),(4,'2026-02-10');
CREATE TABLE events (user_id int, event_ts timestamp);
INSERT INTO events VALUES
  (1,'2026-01-04'),(1,'2026-02-11'),(1,'2026-03-01'),
  (2,'2026-01-21'),(3,'2026-02-03'),(3,'2026-03-05'),(4,'2026-04-01');

CREATE TABLE customer_raw (customer_id int, email text, city text, updated_at timestamp, ingest_id int);
INSERT INTO customer_raw VALUES
  (1,'ranga@x.com','Pune',  '2026-01-01',1),(1,'ranga@x.com','Pune',  '2026-01-01',1),  -- exact duplicate
  (1,'ranga@x.com','Mumbai','2026-02-01',2),                                       -- newer version
  (2,'meena@x.com','Delhi', '2026-01-05',1),
  (3,'vinod@x.com','Chennai','2026-01-07',1),(3,'vinod.k@x.com','Chennai','2026-01-07',2); -- same ts, tiebreak

CREATE TABLE order_header (order_id int PRIMARY KEY, order_amount numeric(10,2));
INSERT INTO order_header VALUES (1,100),(2,200),(3,50);
CREATE TABLE shipments (shipment_id int, order_id int, carrier text, shipped_at timestamp);
INSERT INTO shipments VALUES (1,1,'DHL','2026-10-01'),(2,1,'DHL','2026-10-02'),(3,2,'UPS','2026-10-01');

CREATE TABLE daily_sales (sale_date date, region text, amount numeric(10,2));
INSERT INTO daily_sales VALUES
  ('2026-10-01','N',100),('2026-10-02','N',120),('2026-10-04','N',80),  -- 10-03 missing
  ('2026-10-05','N',90),('2026-10-01','S',50),('2026-10-02','S',55);

CREATE TABLE contacts (customer_id int, email text, updated_at date);
INSERT INTO contacts VALUES
  (14,'yashu@x.com','2026-02-02'),(27,'yashu@x.com','2026-02-19'),(31,'yashu@x.com','2026-02-24'),
  (9,'anada@x.com','2026-02-11'),(18,'anada@x.com','2026-02-11'),   -- tie on updated_at
  (22,'raja@x.com','2026-02-07'),
  (40,NULL,'2026-02-15'),(41,NULL,'2026-02-16');                     -- missing emails are never duplicates

CREATE TABLE app_events (user_id int, event_type text, event_ts timestamp);
INSERT INTO app_events VALUES
  (1,'login','2026-10-01 09:00'),(1,'view','2026-10-01 09:05'),(1,'view','2026-10-01 09:07'),
  (1,'add_to_cart','2026-10-01 09:10'),(1,'purchase','2026-10-01 09:15'),
  (2,'login','2026-10-02 10:00'),(2,'view','2026-10-02 10:02'),(2,'view','2026-10-02 10:04'),
  (2,'add_to_cart','2026-10-02 10:06'),
  (3,'login','2026-10-03 11:00'),(3,'purchase','2026-10-03 11:01'),   -- bought without a cart step
  (4,'login','2026-10-04 12:00');
CREATE TABLE app_users (user_id int PRIMARY KEY, signup_date date);
INSERT INTO app_users VALUES (1,'2026-09-28'),(2,'2026-09-29'),(3,'2026-09-30'),
                             (4,'2026-10-01'),(5,'2026-10-02');      -- user 5 never came back

CREATE TABLE sellers (seller_id int, signup_date date);
INSERT INTO sellers VALUES
  (1,'2021-04-11'),(2,'2021-09-30'),
  (3,'2022-01-05'),(4,'2022-03-17'),(5,'2022-06-21'),(6,'2022-08-02'),(7,'2022-12-12'),
  (8,'2024-02-14'),(9,'2024-05-01'),(10,'2024-07-19');                -- no sellers joined in 2023

CREATE TABLE product_sales (region text, product text, revenue numeric(10,2));
INSERT INTO product_sales VALUES
  ('East','Router',700),('East','Router',300),('East','Switch',450),
  ('East','Cable',450),('East','Adapter',120),('East','Mouse',30),
  ('West','Laptop',1200),('West','Monitor',650),('West','Dock',400);

CREATE TABLE visits (visit_date date, amount numeric(10,2));
INSERT INTO visits VALUES
  ('2026-10-01',40),('2026-10-02',55),('2026-10-03',65),('2026-10-04',80),
  ('2026-10-06',75),                                                  -- 5 October: the shop was closed
  ('2026-10-07',110),('2026-10-08',90),('2026-10-08',15),('2026-10-09',70);

CREATE TABLE person (id int PRIMARY KEY, email text);
INSERT INTO person VALUES
  (11,'kavi@x.com'),(12,'manoj@x.com'),(15,'kavi@x.com'),(19,'nishanth@x.com'),
  (23,'manoj@x.com'),(30,'Kavi@x.com');                  -- differs only in case

CREATE TABLE ride_users (user_id int PRIMARY KEY, banned text, role text);
INSERT INTO ride_users VALUES
  (21,'No','client'),(22,'Yes','client'),(23,'No','client'),(24,'No','client'),
  (31,'No','driver'),(32,'Yes','driver'),(33,'No','driver');   -- driver 34 is not in the table
CREATE TABLE trips (id int, client_id int, driver_id int, status text, request_date date);
INSERT INTO trips VALUES
  (1,21,31,'completed','2026-10-05'),(2,22,31,'cancelled_by_client','2026-10-05'),
  (3,23,33,'cancelled_by_driver','2026-10-05'),(4,21,33,'completed','2026-10-05'),
  (5,24,32,'cancelled_by_driver','2026-10-06'),(6,23,33,'completed','2026-10-06'),
  (7,24,34,'cancelled_by_client','2026-10-07');

CREATE TABLE parcels (parcel_id int, shipped_on date);
INSERT INTO parcels VALUES
  (1,'2026-01-05'),(2,'2026-01-15'),(3,'2026-01-31'),(4,'2026-02-01'),
  (5,'2026-04-10'),(6,'2026-04-11');                    -- nothing shipped in March

CREATE TABLE web_log (user_id int, event_type text, event_time timestamp);
INSERT INTO web_log VALUES
  (1,'page_load','2026-10-01 08:00'),(1,'page_exit','2026-10-01 08:12'),
  (1,'page_load','2026-10-01 08:30'),(1,'page_exit','2026-10-01 08:48'),
  (1,'page_load','2026-10-02 19:00'),(1,'page_exit','2026-10-02 19:24'),
  (2,'page_load','2026-10-01 23:50'),(2,'page_exit','2026-10-02 00:10'),  -- crosses midnight
  (2,'page_load','2026-10-02 08:00'),(2,'page_exit','2026-10-02 08:30'),
  (3,'page_load','2026-10-01 12:00');                                     -- never exits

CREATE TABLE subscriptions (sub_id int, user_id int, start_date date, end_date date);
INSERT INTO subscriptions VALUES
  (1,101,'2026-01-12','2026-01-28'),    -- cancelled inside its first month
  (2,102,'2026-01-20',NULL),            -- still active
  (3,103,'2026-01-05','2026-03-03'),
  (4,104,'2026-01-25','2026-02-01'),    -- ends exactly on the month-1 boundary: retained
  (5,105,'2026-02-03',NULL),
  (6,106,'2026-02-14','2026-04-30');

CREATE TABLE tasks (task_id int, task_date date, status text);
INSERT INTO tasks VALUES
  (1,'2026-10-10','completed'),(2,'2026-10-11','completed'),(3,'2026-10-12','completed'),
  (4,'2026-10-12','completed'),                      -- second task on the same day
  (5,'2026-10-13','pending'),                        -- not completed: breaks the run
  (6,'2026-10-14','completed'),
  (7,'2026-10-16','completed'),(8,'2026-10-17','completed');

CREATE TABLE scores (id int, score int);
INSERT INTO scores VALUES (1,91),(2,85),(3,91),(4,64),(5,85),(6,70),(7,85);

Pattern: top-N per group. The signal is “Nth highest” or “highest per”.

The sample emp table has a tie on purpose: Nishanth and Meena both earn 180,000 in ENG. A correct “second highest” must return the second distinct salary, which is Manoj’s 190,000.

Approach 1, DENSE_RANK(). The default answer. Ties share a rank and no rank is skipped, so rank 2 is the second distinct value.

-- 2nd highest distinct salary per department
SELECT dept, name, salary
FROM (
  SELECT dept, name, salary,
         DENSE_RANK() OVER (PARTITION BY dept ORDER BY salary DESC) AS rnk
  FROM emp
) t
WHERE rnk = 2
ORDER BY dept, name;

 dept  |  name   |  salary   
-------+---------+-----------
 ENG   | Manoj   | 190000.00
 SALES | Lakshmi |  95000.00
(2 rows)

Approach 2, correlated subquery. For engines or interviewers that rule out window functions. A salary is the Nth highest when exactly N-1 distinct salaries in its department are greater. It re-scans the department for every row, so it is quadratic per group. Say that out loud.

-- Same answer without window functions (correlated subquery)
SELECT e1.dept, e1.name, e1.salary
FROM emp e1
WHERE 1 = (SELECT COUNT(DISTINCT e2.salary)
           FROM emp e2
           WHERE e2.dept = e1.dept AND e2.salary > e1.salary)
ORDER BY e1.dept, e1.name;

 dept  |  name   |  salary   
-------+---------+-----------
 ENG   | Manoj   | 190000.00
 SALES | Lakshmi |  95000.00
(2 rows)

Approach 3, LIMIT ... OFFSET. Fine for the overall Nth value, awkward per department. DISTINCT is what makes ties count once.

-- Overall Nth (N = 3) highest distinct salary
SELECT DISTINCT salary
FROM emp
ORDER BY salary DESC
LIMIT 1 OFFSET 2;

  salary   
-----------
 180000.00
(1 row)

Approach 4, self-join with no ranking at all. The TCS and IBM variant (“without RANK, LIMIT or a subquery”). The highest salary has nothing greater than it, so it never appears on the left side of the join, and MAX of what remains is the second highest.

-- Second highest overall without RANK, LIMIT or a subquery per row
SELECT MAX(e1.salary) AS second_highest
FROM emp e1
JOIN emp e2 ON e2.salary > e1.salary;

 second_highest 
----------------
      190000.00
(1 row)

The trap: ROW_NUMBER() = 2 returns Nishanth or Meena, whichever the engine happens to put second, and never Manoj. It answers “second row”, not “second highest salary”.

Run all three over the same ordering and the difference is visible on the tie:

SELECT dept, name, salary,
       ROW_NUMBER() OVER w AS row_number,
       RANK()       OVER w AS rank,
       DENSE_RANK() OVER w AS dense_rank
FROM emp
WHERE dept = 'ENG'
WINDOW w AS (ORDER BY salary DESC)
ORDER BY salary DESC, name;

 dept |   name   |  salary   | row_number | rank | dense_rank 
------+----------+-----------+------------+------+------------
 ENG  | Ranga    | 250000.00 |          1 |    1 |          1
 ENG  | Manoj    | 190000.00 |          2 |    2 |          2
 ENG  | Meena    | 180000.00 |          4 |    3 |          3
 ENG  | Nishanth | 180000.00 |          3 |    3 |          3
 ENG  | Vinod    | 150000.00 |          5 |    5 |          4
(5 rows)

• ROW_NUMBER() numbers rows 1, 2, 3, 4, 5 with no ties. Use it when you need exactly one row per group: deduplication, latest record.
• RANK() gives ties the same rank and then skips: 1, 2, 3, 3, 5. Competition ranking.
• DENSE_RANK() gives ties the same rank with no gap: 1, 2, 3, 3, 4. “Nth distinct value”.

They agree only when the ORDER BY key is unique within the partition. Once there are ties, DENSE_RANK() = 1 returns several rows per group where ROW_NUMBER() = 1 returns one, so a dedup query rewritten with DENSE_RANK silently keeps duplicates.

Two common versions. “Rank scores without gaps” asks for each distinct score with a gap-free rank, which is DENSE_RANK over DISTINCT values:

-- Rank distinct scores with no gaps
SELECT score, DENSE_RANK() OVER (ORDER BY score DESC) AS rank
FROM (SELECT DISTINCT score FROM scores) s
ORDER BY score DESC;

 score | rank 
-------+------
    91 |    1
    85 |    2
    70 |    3
    64 |    4
(4 rows)

Its RANK vs ROW_NUMBER problem wants all three side by side, with ROW_NUMBER ties broken by name, which makes the numbering deterministic:

-- All three ranks, with ROW_NUMBER ties broken by name
SELECT name, salary,
       RANK()       OVER (ORDER BY salary DESC)       AS rank_val,
       DENSE_RANK() OVER (ORDER BY salary DESC)       AS dense_rank_val,
       ROW_NUMBER() OVER (ORDER BY salary DESC, name) AS row_num
FROM emp
WHERE dept IN ('ENG', 'SALES')
ORDER BY salary DESC, name;

   name   |  salary   | rank_val | dense_rank_val | row_num 
----------+-----------+----------+----------------+---------
 Ranga    | 250000.00 |        1 |              1 |       1
 Manoj    | 190000.00 |        2 |              2 |       2
 Meena    | 180000.00 |        3 |              3 |       3
 Nishanth | 180000.00 |        3 |              3 |       4
 Vinod    | 150000.00 |        5 |              4 |       5
 Raja     | 120000.00 |        6 |              5 |       6
 Vasundra | 120000.00 |        6 |              5 |       7
 Lakshmi  |  95000.00 |        8 |              6 |       8
(8 rows)

Meena and Nishanth tie on salary, so RANK gives both 3 and skips to 5, DENSE_RANK gives both 3 and continues at 4, and ROW_NUMBER orders them by name.

The follow-up is determinism. Look at the output: Nishanth got row number 3 and Meena 4, although they sort equal. That order is arbitrary. In Spark it can change between runs, because partitions arrive in a different order. Always add a tiebreaker column, such as ORDER BY salary DESC, emp_id.

Pattern: deduplication. The signal is “latest record”, “remove” or “flag” duplicates. The core is always ROW_NUMBER() over the key, newest first; what changes is whether you keep, delete or flag the other rows.

customer_raw holds three kinds of duplicate: an exact copy (customer 1 loaded twice), a newer version of the same key (customer 1 moved to Mumbai), and two versions with the same timestamp (customer 3), which need a tiebreaker.

Find them:

SELECT customer_id, COUNT(*) AS copies
FROM customer_raw
GROUP BY customer_id
HAVING COUNT(*) > 1
ORDER BY customer_id;

 customer_id | copies 
-------------+--------
           1 |      3
           3 |      2
(2 rows)

Keep the latest, without changing the table. ingest_id breaks the tie for customer 3:

SELECT customer_id, email, city, updated_at
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY customer_id
                            ORDER BY updated_at DESC, ingest_id DESC) AS rn
  FROM customer_raw
) t
WHERE rn = 1
ORDER BY customer_id;

 customer_id |     email     |  city   |     updated_at      
-------------+---------------+---------+---------------------
           1 | ranga@x.com   | Mumbai  | 2026-02-01 00:00:00
           2 | meena@x.com   | Delhi   | 2026-01-05 00:00:00
           3 | vinod.k@x.com | Chennai | 2026-01-07 00:00:00
(3 rows)

Delete in place. The table has no primary key, so there is nothing to tell two identical rows apart. PostgreSQL’s physical row id, ctid, does that. SQL Server uses a deletable CTE (WITH d AS (... ROW_NUMBER() ...) DELETE FROM d WHERE rn > 1) for the same effect.

-- PostgreSQL: delete in place using the physical row id (ctid),
-- because the table has no primary key to tell the copies apart
DELETE FROM customer_raw c
USING (
  SELECT ctid,
         ROW_NUMBER() OVER (PARTITION BY customer_id
                            ORDER BY updated_at DESC, ingest_id DESC) AS rn
  FROM customer_raw
) d
WHERE c.ctid = d.ctid AND d.rn > 1;
SELECT customer_id, email, city FROM customer_raw ORDER BY customer_id;

 customer_id |     email     |  city   
-------------+---------------+---------
           1 | ranga@x.com   | Mumbai
           2 | meena@x.com   | Delhi
           3 | vinod.k@x.com | Chennai
(3 rows)

In Delta, Iceberg or Hudi, do not delete row by row. Rewrite the affected partitions with the deduplicated result (INSERT OVERWRITE, or MERGE from a deduplicated source), so the change is one atomic commit.

Flag instead of delete. A common variant keeps every row and adds an is_duplicate column. Two rules make it harder than it looks: a missing email is never a duplicate, and a tie on updated_at goes to the smallest ID.

-- Flag duplicates instead of deleting them: the newest row per email is kept
-- (ties go to the smallest customer_id), and a missing email is never a duplicate.
SELECT customer_id, email, updated_at,
       CASE WHEN email IS NULL THEN 0
            WHEN ROW_NUMBER() OVER (PARTITION BY email
                                    ORDER BY updated_at DESC, customer_id) = 1 THEN 0
            ELSE 1 END AS is_duplicate
FROM contacts
ORDER BY customer_id;

 customer_id |    email    | updated_at | is_duplicate 
-------------+-------------+------------+--------------
           9 | anada@x.com | 2026-02-11 |            0
          14 | yashu@x.com | 2026-02-02 |            1
          18 | anada@x.com | 2026-02-11 |            1
          22 | raja@x.com  | 2026-02-07 |            0
          27 | yashu@x.com | 2026-02-19 |            1
          31 | yashu@x.com | 2026-02-24 |            0
          40 |             | 2026-02-15 |            0
          41 |             | 2026-02-16 |            0
(8 rows)

Without the email IS NULL branch, PARTITION BY email puts both NULL emails into one group and flags customer 40, because SQL groups NULLs together even though NULL = NULL is not true. Customers 9 and 18 share a date, and the customer_id tiebreaker keeps 9.

The follow-up: an exact duplicate and a key duplicate are different problems. The first is a DISTINCT. The second needs a rule for which version wins, and that rule belongs to the business, not to you.

Pattern: running totals. The signal is “cumulative”, “balance” or “to date”.

Account 10 has two transactions at exactly the same timestamp. That is where the default window frame gives a wrong running balance.

The default frame is RANGE. With an ORDER BY and no frame clause, the frame is RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW. It includes every peer row with the same txn_ts, so both 09:00 rows show the combined 300:

-- Default frame: RANGE, so rows with the same txn_ts are summed together
SELECT account_id, txn_ts, amount,
       SUM(amount) OVER (PARTITION BY account_id ORDER BY txn_ts) AS balance_range
FROM txns ORDER BY account_id, txn_ts, txn_id;

 account_id |       txn_ts        | amount  | balance_range 
------------+---------------------+---------+---------------
         10 | 2026-10-01 09:00:00 |  500.00 |        300.00
         10 | 2026-10-01 09:00:00 | -200.00 |        300.00
         10 | 2026-10-02 10:00:00 |  -50.00 |        250.00
         20 | 2026-10-01 12:00:00 | 1000.00 |       1000.00
(4 rows)

State the frame as ROWS, and add a tiebreaker. Now each row shows the balance after that transaction. The tiebreaker is txn_id, which records arrival order:

-- Explicit ROWS frame, with arrival order (txn_id) breaking the timestamp tie
SELECT account_id, txn_id, txn_ts, amount,
       SUM(amount) OVER (PARTITION BY account_id
                         ORDER BY txn_ts, txn_id
                         ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS balance
FROM txns ORDER BY account_id, txn_ts, txn_id;

 account_id | txn_id |       txn_ts        | amount  | balance 
------------+--------+---------------------+---------+---------
         10 |      1 | 2026-10-01 09:00:00 |  500.00 |  500.00
         10 |      2 | 2026-10-01 09:00:00 | -200.00 |  300.00
         10 |      3 | 2026-10-02 10:00:00 |  -50.00 |  250.00
         20 |      4 | 2026-10-01 12:00:00 | 1000.00 | 1000.00
(4 rows)

The same in PySpark:

from pyspark.sql import Window, functions as F

w = (Window.partitionBy("account_id")
           .orderBy("txn_ts", F.desc("amount"))
           .rowsBetween(Window.unboundedPreceding, Window.currentRow))
txns.withColumn("balance", F.sum("amount").over(w))

“In input order” needs a column. Interview versions of this pattern often say “when several rows share a day, add them in their input order”. A SQL table has no input order: without an ORDER BY on a real column, the database may return rows in any order. Ask for, or create, a column that records it: an ingestion sequence, a Kafka offset, or an auto-increment ID.

For a balance, debits must already be negative. Say that you would check the sign convention in the source, because a feed that stores debits as positive numbers with a type column needs a CASE first.

Pattern: gaps and islands. The signal is “consecutive”, “streak” or “in a row”. The classic version, “login streaks of three or more days”, is the same query as below with logins instead of orders.

LAG/LEAD for a fixed run length. “Rising three months in a row” means this month beats the last, and the last beat the one before:

-- Users whose usage rose three months in a row
SELECT user_id, month, usage
FROM (
  SELECT user_id, month, usage,
         LAG(usage, 1) OVER (PARTITION BY user_id ORDER BY month) AS prev1,
         LAG(usage, 2) OVER (PARTITION BY user_id ORDER BY month) AS prev2
  FROM usage_monthly
) t
WHERE usage > prev1 AND prev1 > prev2
ORDER BY user_id, month;

 user_id |   month    | usage 
---------+------------+-------
       1 | 2026-03-01 |    15
       2 | 2026-04-01 |    45
(2 rows)

Gaps and islands for any run length. Subtract a row number from the date. Within a run of consecutive days the difference is constant, so it labels the run (the “island”):

-- Customers who ordered on 3 or more consecutive days (gaps and islands)
WITH days AS (
  SELECT DISTINCT customer_id, order_date FROM orders     -- one row per day first
),
grp AS (
  SELECT customer_id, order_date,
         order_date - (ROW_NUMBER() OVER (PARTITION BY customer_id
                                          ORDER BY order_date))::int AS island
  FROM days
)
SELECT customer_id, MIN(order_date) AS run_start, MAX(order_date) AS run_end,
       COUNT(*) AS days
FROM grp
GROUP BY customer_id, island
HAVING COUNT(*) >= 3;

 customer_id | run_start  |  run_end   | days 
-------------+------------+------------+------
           1 | 2026-01-05 | 2026-01-07 |    3
(1 row)

Customer 1 has two orders on 7 January. The SELECT DISTINCT in the first CTE matters: without it, the second order on the same day gets its own row number, the subtraction shifts, and the run splits in two.

LAG needs one comparison per step, so it suits “exactly three”. Gaps and islands works for “three or more” and returns the run boundaries too.

Numbered date ranges. The “contiguous date ranges” variant adds two rules: only completed tasks count, and each run gets a sequential ID and an inclusive length. The status filter goes before the row numbering, so a pending day breaks the run:

-- Consecutive completed dates become one numbered interval
WITH done AS (
  SELECT DISTINCT task_date FROM tasks WHERE status = 'completed'
),
grp AS (
  SELECT task_date,
         task_date - (ROW_NUMBER() OVER (ORDER BY task_date))::int AS island
  FROM done
)
SELECT ROW_NUMBER() OVER (ORDER BY MIN(task_date)) AS project_id,
       MIN(task_date) AS start_date, MAX(task_date) AS end_date,
       MAX(task_date) - MIN(task_date) + 1 AS duration_days
FROM grp
GROUP BY island
ORDER BY start_date;

 project_id | start_date |  end_date  | duration_days 
------------+------------+------------+---------------
          1 | 2026-10-10 | 2026-10-12 |             3
          2 | 2026-10-14 | 2026-10-14 |             1
          3 | 2026-10-16 | 2026-10-17 |             2
(3 rows)

12 October has two completed tasks and counts once, because of the DISTINCT. 13 October is pending, so the run from the 10th ends on the 12th. The duration is end - start + 1, since both endpoints count.

When the source is timestamps, cast to a date first (CAST(login_ts AS date)) and then deduplicate, so two logins on one day count once. In Spark the subtraction is date_sub(login_date, rn); in Snowflake, DATEADD(day, -rn, login_date).

NOT IN is the trap. One NULL in the subquery makes x NOT IN (...) unknown for every row, and the query returns nothing:

-- NOT IN with a NULL in the subquery returns nothing
SELECT COUNT(*) AS rows_returned
FROM (VALUES (1), (2), (3)) c(id)
WHERE id NOT IN (SELECT x FROM (VALUES (1), (NULL)) s(x));

 rows_returned 
---------------
             0
(1 row)

NOT EXISTS is the correct shape and is null-safe:

-- Customers who ordered in January but not in February
SELECT DISTINCT o.customer_id
FROM orders o
WHERE o.order_date >= '2026-01-01' AND o.order_date < '2026-02-01'
  AND NOT EXISTS (
    SELECT 1 FROM orders f
    WHERE f.customer_id = o.customer_id
      AND f.order_date >= '2026-02-01' AND f.order_date < '2026-03-01')
ORDER BY 1;

 customer_id 
-------------
           2
           4
(2 rows)

Without any join, EXCEPT gives the same answer, at the cost of deduplicating both sides:

SELECT customer_id FROM orders WHERE order_date >= '2026-01-01' AND order_date < '2026-02-01'
EXCEPT
SELECT customer_id FROM orders WHERE order_date >= '2026-02-01' AND order_date < '2026-03-01'
ORDER BY 1;

 customer_id 
-------------
           2
           4
(2 rows)

In Spark, left_semi returns left rows that have a match, with no right-hand columns and no duplication when the right side matches several times. left_anti returns left rows with no match:

from pyspark.sql import functions as F

jan = orders.filter(F.col("order_date").between("2026-01-01", "2026-01-31"))
feb_customers = (orders
    .filter(F.col("order_date").between("2026-02-01", "2026-02-28"))
    .select("customer_id").distinct())          # shrink the right side first

jan.join(feb_customers, "customer_id", "left_anti").select("customer_id").distinct()

At 100M rows, both sides shuffle on customer_id. Reduce the right side to distinct keys before the join, and keep the table partitioned by month so each side is a single-partition scan. If the February key set is small enough, F.broadcast(feb_customers) removes the shuffle of the large side entirely.

ROW_NUMBER() ... = 2 alone drops users with only one login. The interviewer will ask about them, so answer it first: left join back to the distinct users.

-- Penultimate login per user, keeping users who have only one
WITH ranked AS (
  SELECT user_id, login_ts,
         ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY login_ts DESC) AS rn
  FROM logins
)
SELECT u.user_id, r.login_ts AS second_latest_login
FROM (SELECT DISTINCT user_id FROM logins) u
LEFT JOIN ranked r ON r.user_id = u.user_id AND r.rn = 2
ORDER BY u.user_id;

 user_id | second_latest_login 
---------+---------------------
       1 | 2026-10-03 10:00:00
       2 | 
       3 | 2026-10-01 07:00:00
(3 rows)

User 2 has one login, so their second-latest login is NULL rather than missing. If two logins share a timestamp, add a tiebreaker to the ORDER BY, for the reason given in question 2.

Pattern: recursive hierarchies. The signal is “all reports under”, “chain of command” or “levels”.

The anchor member selects the root. The recursive member joins children to the rows found so far. The path array does two jobs: it shows the chain, and it is the cycle guard.

WITH RECURSIVE chain AS (
  SELECT emp_id, name, manager_id, 1 AS depth, ARRAY[emp_id] AS path
  FROM emp WHERE manager_id IS NULL
  UNION ALL
  SELECT e.emp_id, e.name, e.manager_id, c.depth + 1, c.path || e.emp_id
  FROM emp e
  JOIN chain c ON e.manager_id = c.emp_id
  WHERE NOT e.emp_id = ANY(c.path)      -- cycle guard
    AND c.depth < 20                    -- depth cap
)
SELECT emp_id, name, depth, path FROM chain ORDER BY path;

 emp_id |   name   | depth |  path   
--------+----------+-------+---------
      1 | Ranga    |     1 | {1}
      2 | Nishanth |     2 | {1,2}
      4 | Vinod    |     3 | {1,2,4}
      9 | Manoj    |     3 | {1,2,9}
      3 | Meena    |     2 | {1,3}
      5 | Vasundra |     2 | {1,5}
      6 | Raja     |     3 | {1,5,6}
      7 | Lakshmi  |     3 | {1,5,7}
      8 | Kavi     |     2 | {1,8}
(9 rows)

Everyone under a given manager. The common variant starts the recursion at the manager’s direct reports instead of at the root, which also keeps the manager out of the result:

-- Everyone under a given manager, directly or indirectly, excluding the manager
WITH RECURSIVE reports AS (
  SELECT emp_id, name, manager_id, 1 AS level
  FROM emp WHERE manager_id = 2                       -- start at Nishanth's direct reports
  UNION ALL
  SELECT e.emp_id, e.name, e.manager_id, r.level + 1
  FROM emp e JOIN reports r ON e.manager_id = r.emp_id
)
SELECT emp_id, name, level FROM reports ORDER BY level, emp_id;

 emp_id | name  | level 
--------+-------+-------
      4 | Vinod |     1
      9 | Manoj |     1
(2 rows)

Nishanth’s direct reports are Vinod and Manoj, and neither manages anyone, so the recursion stops after one level. Starting the anchor at emp_id = 2 instead would include Nishanth himself, which the question usually excludes.

Leaf nodes, the ones nobody reports to, are an anti-join on manager_id:

-- Leaf employees: nobody reports to them
SELECT e.emp_id, e.name
FROM emp e
WHERE NOT EXISTS (SELECT 1 FROM emp r WHERE r.manager_id = e.emp_id)
ORDER BY e.emp_id;

 emp_id |  name   
--------+---------
      3 | Meena
      4 | Vinod
      6 | Raja
      7 | Lakshmi
      8 | Kavi
      9 | Manoj
(6 rows)

Name the two failure modes. A cycle in the data (A manages B, B manages A) recurses forever, which is what the path check and the depth cap stop. And the cost grows with depth, because every level is another join.

Spark 3.5 has no WITH RECURSIVE. The usual answer is a loop in the driver that joins one level at a time and stops when a level comes back empty:

from pyspark.sql import functions as F

level = (emp.filter(F.col("manager_id").isNull())
            .select("emp_id", "name", F.lit(1).alias("depth")))
result = level
for depth in range(2, 21):                       # depth cap guards against cycles
    level = (emp.alias("e")
             .join(level.alias("l"), F.col("e.manager_id") == F.col("l.emp_id"))
             .select("e.emp_id", "e.name", F.lit(depth).alias("depth")))
    if level.isEmpty():                          # no new rows: the tree is done
        break
    result = result.unionByName(level)

+------+--------+-----+
|emp_id|    name|depth|
+------+--------+-----+
|     1|   Ranga|    1|
|     2|Nishanth|    2|
|     3|   Meena|    2|
|     5|Vasundra|    2|
|     4|   Vinod|    3|
|     6|    Raja|    3|
|     9|   Manoj|    3|
+------+--------+-----+

Each iteration adds a join to the plan, so for deep trees localCheckpoint() the result every few levels to keep the lineage short.

Take the latest audit row per account, left join it to accounts, and treat a missing status as active:

WITH latest AS (
  SELECT account_id, status,
         ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY changed_at DESC) AS rn
  FROM account_audit
)
SELECT a.account_id, COALESCE(l.status, 'ACTIVE') AS current_status
FROM accounts a
LEFT JOIN latest l ON l.account_id = a.account_id AND l.rn = 1
WHERE COALESCE(l.status, 'ACTIVE') = 'ACTIVE'
ORDER BY a.account_id;

 account_id | current_status 
------------+----------------
          2 | ACTIVE
          4 | ACTIVE
(2 rows)

Account 1’s latest status is SUSPENDED, account 3’s is CLOSED, account 2 was reactivated, and account 4 has no audit rows at all. Account 4 is the point of the question: an inner join to the audit table drops it, and the “count as active” clause is how the interviewer checks whether you noticed.

Pattern: top-N per group. The signal is “top 3 per”, “highest in each”.

Ask first whether ties should all be returned. With DENSE_RANK(), ENG returns four people for “top 3”, because two share third place:

SELECT dept, name, salary, rnk
FROM (
  SELECT dept, name, salary,
         DENSE_RANK() OVER (PARTITION BY dept ORDER BY salary DESC) AS rnk
  FROM emp
) t
WHERE rnk <= 3
ORDER BY dept, rnk, name;

 dept  |   name   |  salary   | rnk 
-------+----------+-----------+-----
 ENG   | Ranga    | 250000.00 |   1
 ENG   | Manoj    | 190000.00 |   2
 ENG   | Meena    | 180000.00 |   3
 ENG   | Nishanth | 180000.00 |   3
 HR    | Kavi     |  90000.00 |   1
 SALES | Raja     | 120000.00 |   1
 SALES | Vasundra | 120000.00 |   1
 SALES | Lakshmi  |  95000.00 |   2
(8 rows)

Aggregate first, then rank. In the usual version the table holds individual sales, so the ranking is over totals, not rows. Rank in a second step, never in the same SELECT as the GROUP BY unless you mean to:

-- Aggregate first, then rank: top 3 revenue levels per region
WITH totals AS (
  SELECT region, product, SUM(revenue) AS total_revenue
  FROM product_sales GROUP BY region, product
),
ranked AS (
  SELECT *, DENSE_RANK() OVER (PARTITION BY region ORDER BY total_revenue DESC) AS region_rank
  FROM totals
)
SELECT region, product, total_revenue, region_rank
FROM ranked
WHERE region_rank <= 3
ORDER BY region, region_rank, product;

 region | product | total_revenue | region_rank 
--------+---------+---------------+-------------
 East   | Router  |       1000.00 |           1
 East   | Cable   |        450.00 |           2
 East   | Switch  |        450.00 |           2
 East   | Adapter |        120.00 |           3
 West   | Laptop  |       1200.00 |           1
 West   | Monitor |        650.00 |           2
 West   | Dock    |        400.00 |           3
(7 rows)

Router’s two sales total 1,000. Cable and Switch tie at 450 and share rank 2, so East returns four products for “top 3”, and Mouse (30) is fourth by revenue level. Ranking before aggregating would rank the two Router rows separately, as 700 and 300.

If the requirement is “exactly three rows”, use ROW_NUMBER() with a tiebreaker. In PySpark both are one window:

from pyspark.sql import Window, functions as F

w = Window.partitionBy("region").orderBy(F.desc("sales"))
(sales.withColumn("dense_rank", F.dense_rank().over(w))
      .withColumn("row_number", F.row_number().over(w.orderBy(F.desc("sales"), "product")))
      .filter("dense_rank <= 2"))

+------+-------+-----+----------+----------+
|region|product|sales|dense_rank|row_number|
+------+-------+-----+----------+----------+
|     N|     p1|  500|         1|         1|
|     N|     p2|  400|         2|         2|
|     N|     p3|  400|         2|         3|
|     S|     p1|  300|         1|         1|
|     S|     p5|  200|         2|         2|
+------+-------+-----+----------+----------+

Work from evidence, in this order:

1. Compare the plans. Get the plan now and, if you can, from when it was fast. A changed join strategy, a lost index, or a filter that stopped pruning partitions explains most regressions.
2. Check what changed. Data volume, stale statistics (which lead to bad join orders), a new predicate, or a function wrapped around a filter column.
3. Reduce the bytes read. Partition and cluster on filter columns, select only the needed columns, push filters below joins.
4. Fix the joins. Pre-aggregate before joining, filter early, broadcast the small side, and check for row explosion from many-to-many keys (question 82 shows one).

Reading a plan in PostgreSQL:

EXPLAIN (COSTS OFF)
SELECT o.customer_id, SUM(o.amount)
FROM orders o
WHERE o.order_date >= DATE '2026-01-01'
GROUP BY o.customer_id;

                     QUERY PLAN                     
----------------------------------------------------
 HashAggregate
   Group Key: customer_id
   ->  Seq Scan on orders o
         Filter: (order_date >= '2026-01-01'::date)
(4 rows)

A Seq Scan with a Filter on a large table is the first thing to question. Use EXPLAIN (ANALYZE, BUFFERS) to get actual row counts and pages read, and compare estimated rows with actual rows: a large gap means stale statistics, so run ANALYZE orders.

The same check in Spark:

df.explain("formatted")       # physical plan, with PushedFilters on each scan
spark.sql("ANALYZE TABLE sales.orders COMPUTE STATISTICS FOR ALL COLUMNS")

The “it used to be fast” framing is deliberate. The interviewer wants to hear that you start with the plan diff, not with adding indexes.

One query shows most of it:

SELECT NULL = NULL                   AS eq,
       NULL IS NOT DISTINCT FROM NULL AS null_safe_eq,
       COUNT(*)                       AS count_star,
       COUNT(x)                       AS count_col,
       AVG(x)                         AS avg_col
FROM (VALUES (10), (NULL), (20)) t(x);

 eq | null_safe_eq | count_star | count_col |       avg_col       
----+--------------+------------+-----------+---------------------
    | t            |          3 |         2 | 15.0000000000000000
(1 row)

• Comparison: NULL = NULL is NULL (the blank in eq), not true, so a WHERE drops the row. Use IS NULL, IS NOT DISTINCT FROM, Spark’s <=>, or eqNullSafe in PySpark.
• Aggregates: COUNT(*) counts 3 rows, COUNT(x) counts 2, and AVG(x) is 15, not 10, because the null is skipped rather than treated as zero.
• Grouping: GROUP BY puts all nulls into one group.

Joins: null keys never match, even with each other:

SELECT COUNT(*) AS matched_rows
FROM (VALUES (NULL::int), (1)) a(k)
JOIN (VALUES (NULL::int), (1)) b(k) ON a.k = b.k;

 matched_rows 
--------------
            1
(1 row)

In PySpark the null-safe version does match them:

a.join(b, a.k == b.k).count()             # 1
a.join(b, a.k.eqNullSafe(b.k)).count()    # 2

equi-join rows    : 1
null-safe join rows: 2

WHERE vs HAVING. WHERE filters rows before grouping, HAVING filters groups after:

SELECT dept, COUNT(*) AS staff, AVG(salary)::int AS avg_salary
FROM emp
WHERE hire_date >= '2017-01-01'          -- filters rows before grouping
GROUP BY dept
HAVING COUNT(*) >= 2                     -- filters groups after
ORDER BY dept;

 dept  | staff | avg_salary 
-------+-------+------------
 ENG   |     3 |     173333
 SALES |     3 |     111667
(2 rows)

CTE vs subquery. Most engines produce the same plan. A CTE is more readable and can be referenced twice. PostgreSQL 12 and later inline a CTE unless you write AS MATERIALIZED, so a CTE is not automatically computed once.

TRUNCATE vs DELETE. DELETE FROM t WHERE ... removes chosen rows, fires row triggers, and is fully logged. TRUNCATE t removes all rows by deallocating storage, with no WHERE. In PostgreSQL both are transactional and can be rolled back; in some other engines TRUNCATE cannot.

Index.

CREATE INDEX idx_orders_customer_date ON orders (customer_id, order_date);

A sorted structure that turns a scan into a seek. It costs write throughput and storage, and a composite index helps a filter on its leading column, not on order_date alone.

Materialized view.

CREATE MATERIALIZED VIEW daily_revenue AS
SELECT order_date, SUM(amount) AS revenue FROM orders GROUP BY order_date;

REFRESH MATERIALIZED VIEW daily_revenue;

A stored query result that trades freshness for read cost. On a lakehouse the “index” question becomes partitioning, clustering and file-level statistics, because Parquet tables have no B-tree indexes.

Define the baseline before writing SQL. Here it is the average month over the six months before the 90-day window, counting months with no purchases as zero. Then “exceeds by 50%” means the last 90 days are more than 1.5 × (3 × average month).

-- Customers whose last-90-day spend is more than 50% above their usual
-- 90-day spend, where "usual" is the average month over the 6 months before
-- the window, counting months with no purchases as zero.
WITH params AS (SELECT DATE '2026-02-28' AS as_of),
months AS (
  SELECT generate_series(DATE '2025-06-01', DATE '2025-11-01', INTERVAL '1 month')::date AS m
),
monthly AS (
  SELECT customer_id, DATE_TRUNC('month', order_date)::date AS m, SUM(amount) AS total
  FROM orders GROUP BY 1, 2
),
baseline AS (
  SELECT c.customer_id, AVG(COALESCE(mo.total, 0)) AS avg_month
  FROM (SELECT DISTINCT customer_id FROM orders) c
  CROSS JOIN months m
  LEFT JOIN monthly mo ON mo.customer_id = c.customer_id AND mo.m = m.m
  GROUP BY c.customer_id
),
recent AS (
  SELECT customer_id, SUM(amount) AS last_90d
  FROM orders, params
  WHERE order_date > as_of - 90 AND order_date <= as_of
  GROUP BY customer_id
)
SELECT r.customer_id, r.last_90d, ROUND(b.avg_month, 2) AS avg_month,
       ROUND(b.avg_month * 3 * 1.5, 2) AS threshold
FROM recent r JOIN baseline b USING (customer_id)
WHERE b.avg_month > 0                       -- no baseline: a new customer, not a spike
  AND r.last_90d > 1.5 * b.avg_month * 3
ORDER BY 1;

 customer_id | last_90d | avg_month | threshold 
-------------+----------+-----------+-----------
           2 |   300.00 |      8.33 |     37.50
(1 row)

Customer 2 spent 50 in the baseline (8.33 a month) and 300 in the last 90 days, so it is flagged. Customer 1 averaged 75 a month, giving a threshold of 337.50, and spent 270, so it is not.

The trap is the zero months. Months with no rows do not exist in a GROUP BY, so averaging only the months that have orders overstates the baseline. Customer 1 ordered in three of the six baseline months. Without the calendar CTE its average would be 150, not 75. Generate a month calendar and LEFT JOIN to it, as above.

Also decide what to do with customers who have no baseline at all. This query excludes them (avg_month > 0), because a new customer is not a spike.

from collections import Counter, defaultdict

def dedupe_keep_order(items):
    return list(dict.fromkeys(items))          # dicts keep insertion order

def duplicates_by_key(records, key):
    groups = defaultdict(list)
    for r in records:
        groups[r[key]].append(r)
    return {k: v for k, v in groups.items() if len(v) > 1}

def consecutive_ranges(nums):
    out = []
    for x in sorted(set(nums)):
        if out and x == out[-1][1] + 1:
            out[-1][1] = x                     # extend the current range
        else:
            out.append([x, x])                 # start a new one
    return [f"{a}-{b}" if a != b else str(a) for a, b in out]

print(dedupe_keep_order([3, 1, 3, 2, 1]))
print(Counter("mississippi").most_common(2))
emps = [{"id": 1, "name": "A"}, {"id": 2, "name": "B"}, {"id": 1, "name": "A2"}]
print(duplicates_by_key(emps, "id"))
print(consecutive_ranges([1, 2, 3, 7, 8, 10]))

[3, 1, 2]
[('i', 4), ('s', 4)]
{1: [{'id': 1, 'name': 'A'}, {'id': 1, 'name': 'A2'}]}
['1-3', '7-8', '10']

Give the complexity for each. All are O(n), except the ranges, which are O(n log n) because of the sort. most_common returns ties in the order the elements were first seen, so i comes before s here.

import heapq
from collections import Counter, OrderedDict

def merge_intervals(iv):                       # O(n log n)
    out = []
    for s, e in sorted(iv):
        if out and s <= out[-1][1]:
            out[-1][1] = max(out[-1][1], e)
        else:
            out.append([s, e])
    return out

def longest_unique_substring(s):               # O(n) sliding window
    last, start, best = {}, 0, 0
    for i, ch in enumerate(s):
        if ch in last and last[ch] >= start:
            start = last[ch] + 1
        last[ch] = i
        best = max(best, i - start + 1)
    return best

def longest_consecutive(nums):                 # O(n)
    s, best = set(nums), 0
    for x in s:
        if x - 1 not in s:                     # only start counting at a run's start
            n = 1
            while x + n in s:
                n += 1
            best = max(best, n)
    return best

def first_non_repeating(s):                    # O(n)
    c = Counter(s)
    return next((ch for ch in s if c[ch] == 1), None)

def pairs_with_sum(nums, target):              # O(n)
    seen, pairs = set(), set()
    for x in nums:
        if target - x in seen:
            pairs.add((min(x, target - x), max(x, target - x)))
        seen.add(x)
    return sorted(pairs)

def top_k_frequent(items, k):                  # O(n log k)
    return [x for x, _ in heapq.nlargest(k, Counter(items).items(), key=lambda kv: kv[1])]

class FirstUnique:
    """First non-repeating element of a stream, O(1) per event."""
    def __init__(self):
        self.candidates = OrderedDict()
        self.repeated = set()
    def add(self, x):
        if x in self.repeated:
            return
        if x in self.candidates:
            del self.candidates[x]
            self.repeated.add(x)
        else:
            self.candidates[x] = None
    def first(self):
        return next(iter(self.candidates), None)

print(merge_intervals([[1, 3], [2, 6], [8, 10], [9, 12]]))
print(longest_unique_substring("abcabcbb"))
print(longest_consecutive([100, 4, 200, 1, 3, 2]))
print(first_non_repeating("swiss"))
print(pairs_with_sum([2, 7, 4, 5, 3, 6], 9))
print(top_k_frequent(["a", "b", "a", "c", "b", "a"], 2))
fu = FirstUnique()
for x in [5, 3, 5, 7, 3]:
    fu.add(x)
print("first unique in stream:", fu.first())

[[1, 6], [8, 12]]
3
4
w
[(2, 7), (3, 6), (4, 5)]
['a', 'b']
first unique in stream: 7

For “first non-repeating element in a stream” (PayPal), the follow-up is memory. FirstUnique keeps only the current candidates in order plus a set of values already seen twice, so each event is O(1) and the stream itself is never stored.

Flatten nested JSON. Recursion over dicts and lists, joining keys with a separator:

def flatten(obj, parent="", sep="."):
    out = {}
    if isinstance(obj, dict):
        for k, v in obj.items():
            out.update(flatten(v, f"{parent}{sep}{k}" if parent else k, sep))
    elif isinstance(obj, list):
        for i, v in enumerate(obj):
            out.update(flatten(v, f"{parent}{sep}{i}" if parent else str(i), sep))
    else:
        out[parent] = obj
    return out

order = {"id": 7, "customer": {"name": "Ranga", "address": {"city": "Pune"}},
         "items": [{"sku": "A", "qty": 2}, {"sku": "B", "qty": 1}]}
print(flatten(order))

def explode_items(order):
    """The other choice for lists: one output row per list element."""
    base = {k: v for k, v in flatten(order).items() if not k.startswith("items.")}
    return [{**base, **{f"item.{k}": v for k, v in item.items()}} for item in order["items"]]

for row in explode_items(order):
    print(row)

{'id': 7, 'customer.name': 'Ranga', 'customer.address.city': 'Pune', 'items.0.sku': 'A', 'items.0.qty': 2, 'items.1.sku': 'B', 'items.1.qty': 1}
{'id': 7, 'customer.name': 'Ranga', 'customer.address.city': 'Pune', 'item.sku': 'A', 'item.qty': 2}
{'id': 7, 'customer.name': 'Ranga', 'customer.address.city': 'Pune', 'item.sku': 'B', 'item.qty': 1}

Ask which one the interviewer wants. Indexed keys keep one row per order; exploding gives one row per item. The row count changes, and so does every sum downstream.

Large CSV and top routes without pandas. Stream with csv.DictReader, which holds one line at a time, and keep only the aggregate in memory. The test file has 200,000 trips:

import csv, heapq
from collections import defaultdict

def top_routes(path, k=3):
    revenue = defaultdict(float)
    with open(path, newline="") as fh:          # one line in memory at a time
        for row in csv.DictReader(fh):
            revenue[(row["origin"], row["dest"])] += float(row["fare"])
    return heapq.nlargest(k, revenue.items(), key=lambda kv: kv[1])

for route, rev in top_routes(path):
    print(route, round(rev, 2))

('A', 'X') 348084.34
('B', 'W') 347387.78
('C', 'X') 345089.67

Memory is proportional to the number of distinct routes, not to the file. For anything that needs a sort or a join across the whole file, say you would hand it to DuckDB or Spark rather than write an external sort by hand.

Decorator. A function that takes a function and returns a wrapped one. A retry decorator is the example worth giving, because it is what data engineers actually write:

import time
from functools import wraps

def retry(times=3, base_delay=0.01, exceptions=(Exception,)):
    def deco(fn):
        @wraps(fn)                             # keep fn's name and docstring
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions:
                    if attempt == times:
                        raise                  # re-raise the last error unchanged
                    time.sleep(base_delay * 2 ** (attempt - 1))
        return wrapper
    return deco

calls = {"n": 0}

@retry(times=3, exceptions=(ConnectionError,))
def flaky():
    """Fails twice, then succeeds."""
    calls["n"] += 1
    if calls["n"] < 3:
        raise ConnectionError("transient")
    return "ok"

print(flaky(), "after", calls["n"], "calls; name kept:", flaky.__name__,
      "| doc kept:", flaky.__doc__)

ok after 3 calls; name kept: flaky | doc kept: Fails twice, then succeeds.

Retry only the exceptions that can succeed on retry. A ValueError from bad data will fail the same way three times.

Threads vs processes. CPython’s GIL lets one thread run Python bytecode at a time, so threads help I/O-bound work (waiting on a network or disk releases the GIL) and processes help CPU-bound work:

from concurrent.futures import ThreadPoolExecutor

def io_task(_):
    time.sleep(0.2)                             # stands in for a network call
    return 1

t0 = time.perf_counter()
with ThreadPoolExecutor(8) as ex:
    list(ex.map(io_task, range(8)))
print(f"8 x 0.2s sleeps on 8 threads finished in under 0.5s: {time.perf_counter() - t0 < 0.5}")

8 x 0.2s sleeps on 8 threads finished in under 0.5s: True

For CPU-bound work, ProcessPoolExecutor has the same interface; question 99 uses it.

Mutability. Lists, dicts and sets are mutable; tuples, strings and frozensets are not. Question 93 shows the classic bug, a mutable default argument, and question 94 shows shallow versus deep copies.

Exceptions. Catch the narrowest exception you can handle, use finally or a with block for cleanup, and re-raise with raise (not raise e) to keep the original traceback.

spark-submit \
  --master yarn --deploy-mode cluster \
  --num-executors 10 --executor-cores 4 --executor-memory 8g \
  --conf spark.sql.shuffle.partitions=400 \
  daily_orders.py --date 2026-10-01

1. spark-submit starts the driver. In cluster deploy mode it runs inside the cluster (on YARN, in the ApplicationMaster container); in client mode it runs on the machine you submitted from.
2. The driver creates the SparkSession and asks the cluster manager (YARN, Kubernetes or standalone) for executors.
3. Your code builds a logical plan. Nothing executes yet.
4. An action (count, collect, write) hands the plan to Catalyst, which analyses it, optimises it and picks a physical plan.
5. The DAG scheduler cuts the physical plan into stages at every shuffle boundary.
6. The task scheduler sends one task per partition of each stage to executor cores.
7. Map-side tasks write shuffle files to local disk; the next stage’s tasks fetch them over the network.
8. Results go back to the driver (collect) or are written to storage by the executors (write).

The follow-up is always “where does it break?”: the driver on collect() or a large broadcast, executors at a shuffle or on a skewed partition, and the cluster manager when it cannot grant the containers you asked for (the application then sits in ACCEPTED on YARN, or pods stay Pending on Kubernetes).

Do not guess: ask Spark. The status tracker reports every job and stage a job group created. The CSV here is written as 4 files.

from pyspark.sql import functions as F

def count_work(group, fn):
    sc.setJobGroup(group, group)
    fn()
    jobs = sc.statusTracker().getJobIdsForGroup(group)
    stages = []
    for j in sorted(jobs):
        for s in sc.statusTracker().getJobInfo(j).stageIds:
            info = sc.statusTracker().getStageInfo(s)
            if info is not None:
                stages.append((j, s, info.numTasks))
    return sorted(jobs), stages

def pipeline():
    df = spark.read.option("header", True).option("inferSchema", True).csv(csv_path)
    (df.repartition(2)
       .filter(F.col("amount") > 10)
       .select("customer_id", "amount")
       .groupBy("customer_id").agg(F.sum("amount"))
       .collect())

jobs, stages = count_work("seq", pipeline)                    # AQE on (the default)
spark.conf.set("spark.sql.adaptive.enabled", "false")
jobs, stages = count_work("seq_noaqe", pipeline)              # AQE off

jobs: 5
  job 0  stage 56  tasks 1
  job 1  stage 57  tasks 4
  job 2  stage 58  tasks 4
  job 3  stage 60  tasks 2
  job 3  stage 59  tasks 4
  job 4  stage 63  tasks 1
  job 4  stage 61  tasks 4
  job 4  stage 62  tasks 2
with AQE disabled, jobs: 3
  job 0  stage 64  tasks 1
  job 1  stage 65  tasks 4
  job 2  stage 66  tasks 4
  job 2  stage 67  tasks 2
  job 2  stage 68  tasks 200

Read it with AQE off first, because that is the textbook answer:

• Job 0, 1 task: reading the header. With header=true, Spark reads the first line to get the column names.
• Job 1, 4 tasks: inferSchema. A full pass over all 4 files to work out the types. This is the job the question is designed to catch.
• Job 2, the collect(), 3 stages: the scan (4 tasks, one per file split), the stage after repartition(2) (2 tasks), and the stage after the groupBy shuffle (spark.sql.shuffle.partitions, 200 tasks).

With AQE on, the job count rises to 5. AQE runs each shuffle stage as its own job so it can look at the real sizes before planning the next one, and it coalesced the final 200 partitions into 1. Both answers are correct for their settings. Say which one you are describing.

Passing an explicit schema removes jobs 0 and 1 entirely, which is also the answer to “how would you make this faster”.

A narrow transformation (filter, select, withColumn) computes each output partition from one input partition. A wide one (groupBy, join, distinct, repartition) needs rows from many partitions, so it forces a shuffle, and the shuffle is a stage boundary. In the plan, the shuffle is the Exchange node:

df = spark.range(0, 1000).withColumn("k", F.col("id") % 10)
df.filter("id > 5").select("k").groupBy("k").count().explain()

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[k#257L], functions=[count(1)])
   +- Exchange hashpartitioning(k#257L, 200), ENSURE_REQUIREMENTS, [plan_id=2145]
      +- HashAggregate(keys=[k#257L], functions=[partial_count(1)])
         +- Project [(id#255L % 10) AS k#257L]
            +- Filter (id#255L > 5)
               +- Range (0, 1000, step=1, splits=4)

Everything below Exchange (range, filter, project, partial count) runs in one stage without moving data. The Exchange hashpartitioning(k, 200) moves rows so that each key lands in one partition. Note the two HashAggregate nodes: Spark counts within each partition before the shuffle (partial_count), so only 10 partial results per partition cross the network, not 1,000 rows.

Transformations vs actions. Transformations only add to the plan; actions (collect, count, write, show) run it. Laziness is what lets Catalyst see the whole query, push the filter down to the scan, and drop unused columns before anything runs.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("orders").getOrCreate()   # the entry point
sc = spark.sparkContext                                         # the lower-level handle

# RDD: Python objects and lambdas. Spark cannot see inside them.
rdd = sc.parallelize([("A", 10), ("B", 5), ("A", 7)])
totals_rdd = rdd.reduceByKey(lambda a, b: a + b)

# DataFrame: named, typed columns. Catalyst can optimise this.
df = spark.createDataFrame(rdd, "customer STRING, amount INT")
totals_df = df.groupBy("customer").sum("amount")

An RDD is a distributed collection of objects with no schema. Spark cannot reorder or prune what it cannot see, and in Python every row is serialized to a Python worker. A DataFrame carries a schema, so Catalyst prunes columns and pushes filters, and Tungsten stores rows in a compact binary format off the Python heap. A Dataset (Scala and Java only) is a typed DataFrame, Dataset[Order]; in Python only the DataFrame exists. Question 122 shows the one cost of the typed API.

SparkSession (since 2.0) is the single entry point for SQL, DataFrames and the catalog. It wraps a SparkContext, the lower-level connection to the cluster that RDDs and accumulators still use.

Catalyst turns a query into a parsed plan, resolves names against the catalog (the analysed plan), applies rule-based rewrites such as predicate pushdown, constant folding and column pruning (the optimised plan), then chooses a physical plan. Whole-stage code generation compiles chains of operators into one JVM function; the *(1) prefixes in a plan mark them.

Predicate pushdown moves a filter into the data source, so less data is read at all:

spark.read.parquet(pq).filter("customer_id = 7 AND amount > 500") \
     .select("order_id", "amount").explain()

== Physical Plan ==
*(1) Project [order_id#279L, amount#281]
+- *(1) Filter (((isnotnull(customer_id#280L) AND isnotnull(amount#281)) AND (customer_id#280L = 7)) AND (amount#281 > 500.0))
   +- *(1) ColumnarToRow
      +- FileScan parquet [order_id#279L,customer_id#280L,amount#281] Batched: true, DataFilters: [isnotnull(customer_id#280L), isnotnull(amount#281), (customer_id#280L = 7), (amount#281 > 500.0)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/qa_spark/orders_parquet], PartitionFilters: [], PushedFilters: [IsNotNull(customer_id), IsNotNull(amount), EqualTo(customer_id,7), GreaterThan(amount,500.0)], ReadSchema: struct<order_id:bigint,customer_id:bigint,amount:double>

Three things to point at in that scan node:

• PushedFilters: the predicates handed to the Parquet reader, which uses row-group min/max statistics to skip row groups that cannot match.
• ReadSchema: only three columns are read, although the file may have more. That is column pruning.
• The Filter above the scan stays. Pushdown skips row groups, not individual rows, so Spark still filters the rows it reads.

A JDBC source pushes the same filters into the database’s WHERE clause.

File scans split files into chunks of at most spark.sql.files.maxPartitionBytes (128 MB by default), packing small files together using spark.sql.files.openCostInBytes. One 337 MB Parquet file, read twice:

for mb in (128, 32):
    spark.conf.set("spark.sql.files.maxPartitionBytes", f"{mb}m")
    print(f"maxPartitionBytes={mb}m -> partitions:", spark.read.parquet(big).rdd.getNumPartitions())

file size: 337 MB
maxPartitionBytes=128m -> partitions: 4
maxPartitionBytes=32m -> partitions: 11
shuffle partitions default: 200

Four partitions at 128 MB, not three. The split size also takes spark.sql.files.minPartitionNum (the default parallelism, 4 here) into account, so Spark makes splits smaller to keep every core busy. The file count alone does not set the partition count.

Shuffles produce spark.sql.shuffle.partitions partitions (200), which AQE then coalesces toward spark.sql.adaptive.advisoryPartitionSizeInBytes (64 MB). Question 128 shows 200 becoming 1.

Explicitly, repartition(n) and coalesce(n) set it.

A useful target is 100 to 200 MB of input per task. Much smaller and scheduling overhead dominates; much larger and tasks spill and run long.

Start from 8 equal partitions of 100,000 rows and look at the row counts per partition:

base = spark.range(0, 100_000, numPartitions=8)
sizes = lambda d: d.rdd.glom().map(len).collect()

print("original        :", sizes(base))
print("coalesce(3)     :", sizes(base.coalesce(3)))
print("repartition(3)  :", sizes(base.repartition(3)))
r = base.repartitionByRange(3, "id")
print("byRange(3)      :", sizes(r))

original        : [12500, 12500, 12500, 12500, 12500, 12500, 12500, 12500]
coalesce(3)     : [25000, 37500, 37500]
repartition(3)  : [33335, 33332, 33333]
byRange(3)      : [33498, 33080, 33422]
byRange min/max : [(0, 33497), (33498, 66577), (66578, 99999)]

• coalesce(3) merges existing partitions without a shuffle, so the result is uneven (25,000 against 37,500). It can only reduce the partition count.
• repartition(3) does a full shuffle (round-robin, or hash when you pass columns), so the result is even. It can increase or decrease.
• repartitionByRange(3, "id") samples the column to choose boundaries, then shuffles, so each partition holds a contiguous, non-overlapping range. Use it before writing sorted output, so files have tight min/max statistics.

The trap: coalesce(1) before a write does not just produce one file. With no shuffle to separate them, it pulls the whole upstream stage down to one task, so the entire computation runs single-threaded. repartition(1) keeps upstream parallelism and pays for one shuffle instead.

The driver collects the small side, builds a hash table, and ships it to every executor, so the large side joins where it sits with no shuffle. Spark picks it when a side’s estimated size is below spark.sql.autoBroadcastJoinThreshold (10 MB):

facts = spark.range(0, 1_000_000).withColumn("cust", F.col("id") % 1000)
dim = spark.range(0, 1000).withColumnRenamed("id", "cust").withColumn("segment", F.lit("gold"))

facts.join(dim, "cust").explain()                    # automatic

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
facts.join(dim, "cust").explain()                    # broadcast disabled

facts.join(F.broadcast(dim), "cust").explain()       # hint overrides the threshold

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [cust#313L, id#311L, gold AS segment#320]
   +- BroadcastHashJoin [cust#313L], [cust#318L], Inner, BuildRight, false
      :- Project [id#311L, (id#311L % 1000) AS cust#313L]
      :  +- Filter isnotnull((id#311L % 1000))
      :     +- Range (0, 1000000, step=1, splits=4)
      +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=2309]
         +- Project [id#316L AS cust#318L]
            +- Range (0, 1000, step=1, splits=4)


--- threshold -1, AQE off:
== Physical Plan ==
*(5) Project [cust#313L, id#311L, gold AS segment#320]
+- *(5) SortMergeJoin [cust#313L], [cust#318L], Inner
   :- *(2) Sort [cust#313L ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(cust#313L, 200), ENSURE_REQUIREMENTS, [plan_id=2347]
   :     +- *(1) Project [id#311L, (id#311L % 1000) AS cust#313L]
   :        +- *(1) Filter isnotnull((id#311L % 1000))
   :           +- *(1) Range (0, 1000000, step=1, splits=4)
   +- *(4) Sort [cust#318L ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(cust#318L, 200), ENSURE_REQUIREMENTS, [plan_id=2353]
         +- *(3) Project [id#316L AS cust#318L]
            +- *(3) Range (0, 1000, step=1, splits=4)


--- explicit hint:
== Physical Plan ==
*(2) Project [cust#313L, id#311L, gold AS segment#320]
+- *(2) BroadcastHashJoin [cust#313L], [cust#318L], Inner, BuildRight, false
   :- *(2) Project [id#311L, (id#311L % 1000) AS cust#313L]
   :  +- *(2) Filter isnotnull((id#311L % 1000))
   :     +- *(2) Range (0, 1000000, step=1, splits=4)
   +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=2406]
      +- *(1) Project [id#316L AS cust#318L]
         +- *(1) Range (0, 1000, step=1, splits=4)

With the threshold disabled, the same join becomes a SortMergeJoin with an Exchange and a Sort on both sides: the million-row side is shuffled too. The hint brings back BroadcastHashJoin regardless of the threshold.

Avoid it when the small side is big once decompressed into a hash table, because it must fit in driver memory and in every executor. A forced hint on a wrong size estimate is a common cause of driver OOM, and of spark.sql.broadcastTimeout (300 seconds) failures.

When neither side is small: filter and project both sides first, pre-aggregate if the join is followed by a groupBy, bucket both tables on the join key for joins you run repeatedly (question 27), and let AQE switch to a broadcast at runtime if a side turns out small after filtering.

Partitioning splits a table into directories by column value, so queries skip whole directories. It suits low-cardinality filter columns such as date:

df.write.partitionBy("order_date").parquet("s3://lake/orders")   # .../order_date=2026-10-01/

Bucketing hashes a column into a fixed number of files. Two tables bucketed the same way on the join key can join with no shuffle. It suits high-cardinality join keys:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")   # rule out broadcast

(orders.write.bucketBy(8, "cust").sortBy("cust").saveAsTable("orders_b"))
(customers.write.bucketBy(8, "cust").sortBy("cust").saveAsTable("customers_b"))

plan = spark.table("orders_b").join(spark.table("customers_b"), "cust") \
            ._jdf.queryExecution().executedPlan().toString()
print("Exchange in bucketed join plan:", "Exchange" in plan)

Exchange in bucketed join plan: False
Exchange when one side is not bucketed: True

Both tables bucketed on cust into 8 buckets: no Exchange. Join the bucketed table to an unbucketed one and the shuffle comes back. Bucketing needs saveAsTable, because the bucket spec lives in the catalog, and both sides must use the same bucket count.

Broadcasting is a per-query choice for one small table. Bucketing is a layout paid once at write time, for large joins you run every day.

A bloom filter is a compact bit array that answers “definitely not present” or “possibly present”. Built from the join keys of the filtered small side, it lets the large side drop rows that cannot match before the shuffle.

Spark injects one at runtime when the creation side has a selective filter and the application side is large. The size threshold defaults to 10 GB, so the demo lowers it:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold", "1KB")

fact = spark.read.parquet(f"{base}/fact_bloom")                     # 2M rows
vip = spark.read.parquet(f"{base}/dim_bloom").filter(F.col("tier") == "vip")
q = fact.join(vip, "cust")
plan = q._jdf.queryExecution().optimizedPlan().toString()
print("bloom filter injected:", "might_contain" in plan)

bloom filter injected: True
Filter (isnotnull(cust#17L) AND might_contain(scalar-subquery#31 [], xxhash64(cust#17L, 42)))
  +- Aggregate [bloom_filter_agg(xxhash64(cust#20L, 42), 1000000, 8388608, 0, 0) AS bloomFilter#30]

The bloom_filter_agg builds the filter from the VIP customers, and the might_contain filter on the fact side drops non-matching rows during the scan. It is controlled by spark.sql.optimizer.runtime.bloomFilter.enabled, which is true by default since Spark 3.3.

A failed first attempt worth knowing: filtering the dimension on a bare boolean column, .filter("vip"), injected nothing. Spark only builds a filter when the creation side has a predicate it considers selective, such as an equality, IN or LIKE.

Parquet and Delta also store bloom filters per column for point lookups on high-cardinality fields, where min/max statistics cannot prune.

Find where it fails before changing anything:

• Driver OOM (java.lang.OutOfMemoryError in the driver log, often during BroadcastExchange): a broadcast that was too large. Remove the hint, or lower the threshold.
• Executor OOM or a container killed for exceeding memory, on one or two tasks: skew. One key’s rows do not fit in one task. See question 30.
• Every task spilling heavily: too few shuffle partitions for the data size.

Then, in order:

# 1. Shrink both sides before the join
left = big_a.filter("event_date >= '2026-10-01'").select("cust", "amount")
right = big_b.select("cust", "segment").dropDuplicates(["cust"])   # check the key is unique

# 2. More, smaller shuffle partitions, so each task holds less
spark.conf.set("spark.sql.shuffle.partitions", "2000")
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# 3. Check for row explosion before blaming memory
print(left.groupBy("cust").count().orderBy(F.desc("count")).first())
print(right.groupBy("cust").count().filter("count > 1").count())   # many-to-many?

joined = left.join(right, "cust")

A many-to-many key multiplies rows (10,000 × 10,000 matches on one key is 100 million output rows), and no memory setting fixes that. Raising executor memory is the last step, not the first.

In the Spark UI, open the slow stage and read the summary metrics table: compare Max with Median for Duration, Shuffle Read Size and Records. A max many times the median, on a stage right after a shuffle, is skew. Spill on that one task confirms it.

Then find the key. The test data puts 60% of a million events on user_id = 0:

events = (spark.range(0, 1_000_000)
          .withColumn("user_id", F.when(F.col("id") % 10 < 6, F.lit(0))
                                   .otherwise(F.col("id") % 10_000)))

events.groupBy("user_id").count().orderBy(F.desc("count")).show(3)
print(events.repartition(8, "user_id").rdd.glom().map(len).collect())

+-------+------+
|user_id| count|
+-------+------+
|      0|600000|
|     26|   100|
|     29|   100|
+-------+------+
only showing top 3 rows

partition sizes after hash shuffle on user_id (8 partitions):
[49100, 50400, 49300, 52200, 49000, 650900, 48700, 50400]

After a hash shuffle into 8 partitions, one partition holds 650,900 rows and the other seven about 50,000 each. That one partition is the 45-minute task. In real data the hot key is usually a NULL, a default such as 0 or 'UNKNOWN', or one very large customer.

Fix 1, AQE skew join. On by default since Spark 3.2. It splits a shuffle partition that is both 5 times the median and larger than 256 MB, and replicates the matching partition from the other side. These are the settings it uses:

spark.sql.adaptive.enabled = true
spark.sql.adaptive.skewJoin.enabled = true
spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5.0
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 268435456b
spark.sql.adaptive.coalescePartitions.enabled = true
spark.sql.adaptive.advisoryPartitionSizeInBytes = 67108864b

It acts only on sort-merge and shuffled hash joins, and only above both thresholds, so a 200 MB hot partition is left alone. Lower skewedPartitionThresholdInBytes if your partitions are small.

Fix 2, salting. Spread the hot key across N sub-keys on the large side, and replicate the other side N times so every sub-key still finds its match:

N = 8
salted_events = events.withColumn("salt", (F.rand(7) * N).cast("int"))
salted_users = users.crossJoin(spark.range(N).withColumnRenamed("id", "salt"))
joined = salted_events.join(salted_users, ["user_id", "salt"]).drop("salt")

print("salted join rows:", joined.count(), " plain join rows:", events.join(users, "user_id").count())
print(salted_events.filter("user_id = 0").repartition(8, "user_id", "salt").rdd.glom().map(len).collect())

salted join rows: 1000000  plain join rows: 1000000
hot key spread over partitions after salting:
[74936, 225227, 149636, 0, 0, 75260, 0, 74941]

Same row count as the plain join, so the result is correct. The 600,000 hot rows now land in several partitions instead of one. Hash collisions put two salts in the same partition here, which is why salting with N = 8 does not give 8 equal pieces.

Choosing N (HDFC asked this directly): roughly the hot key’s size divided by the target partition size. A 6 GB hot key with a 256 MB target needs about 24. Too small leaves the partition hot; too large multiplies the other side by N for nothing.

Fix 3, when one key is 60% of the data. First ask whether that key should be joined at all. If it is a null or a placeholder, filter it out and union it back. If it is real, join it separately with a broadcast of its single matching row:

hot = events.filter("user_id = 0")
rest = events.filter("user_id <> 0")
out = (rest.join(users, "user_id")
       .unionByName(hot.join(F.broadcast(users.filter("user_id = 0")), "user_id")))

rows: 1000000

The hot rows never shuffle, because the other side of their join is one row.