⚡ PySpark

MapReduce is one map plus one reduce per job. Spark can pipeline many narrow steps in one stage. Spark still writes shuffle files to disk and can spill when memory is full. In production, Spark is not always 100x faster. The big win is reuse and multi-step plans. A single-pass ETL job still spends a lot of time on disk I/O.

Your main() and SparkSession live on the driver. Each executor has cores and memory. One task uses one core at a time and processes one partition. collect() sends data back to the driver, so a huge collect can OOM the driver even when executors are fine. In production, size driver memory for metadata, plans, and small collected results, not for the full dataset.

Before Spark 2, people created SparkContext, SQLContext, and HiveContext separately. SparkSession wraps that. In PySpark you almost always start with SparkSession. RDD code still needs the context, but you get it from the session. In production, do not create many sessions in one app.

The DAGScheduler splits the graph at shuffle boundaries. Each stage is a set of tasks that can pipeline without a shuffle. Each task processes one partition. In production, a long DAG is fine if stages are healthy. A long lineage after many iterations can make recovery slow, which is why checkpoint exists.

filter then select does not run twice. Spark sees both and can push the filter and prune columns. That saves shuffle bytes and memory. The trap in notebooks is calling many actions, which recomputes the same DAG unless you cache. In production, one write action is usually the real trigger.

If you only chain transformations, the cluster is idle. That is normal. groupBy().agg() is still a transformation until you call show or write. collect() and toPandas() pull data to the driver and can OOM. In production, prefer write actions over collecting full results.

Use DataFrames for almost all ETL. Spark can prune columns, push filters, and pick joins. RDDs are for unstructured text or custom per-partition logic that DataFrames cannot express. Python UDFs on DataFrames can fall back toward row-by-row Python work and lose those optimizations. In production, default to DataFrame.

This is why RDDs are "resilient" without 3-way replication of every temp dataset. Short lineage is cheap to replay. Very long lineage, such as many ML iterations, makes recovery expensive. Checkpoint writes data to reliable storage and cuts the chain. Cache keeps data for reuse but does not truncate lineage the same way.

Shuffle cost is serialize, disk write, network, deserialize, and often sort. Wide transformations cause shuffles. Narrow ones do not. In production, look at shuffle read and write bytes in Spark UI. Reduce shuffle by filtering early, broadcasting a small side, using reduceByKey style local aggregation, and avoiding extra repartition.

Narrow steps can pipeline in one stage. Wide steps create a stage boundary. coalesce to fewer partitions is usually narrow. repartition is wide. In production, count shuffle stages, not just lines of code. One extra wide join can dominate runtime and memory.

Too few partitions means huge tasks, less parallelism, and OOM risk. Too many partitions means tiny tasks, scheduler overhead, and many small files on write. A common target is about 100-200 MB per partition after shuffle. AQE can coalesce small shuffle partitions at runtime. In production, size from bytes, not from a lucky number.

repartition("key") hash-partitions for later joins or writes. coalesce(1) makes one task and one file, which is a bottleneck on large data. Coalesce can leave uneven partitions because it does not reshuffle. In production, pick a file count from output size, then coalesce or repartition to that count before write.

Spark can auto-broadcast under spark.sql.autoBroadcastJoinThreshold (often 10 MB). You can force it with F.broadcast(small_df). Broadcasting a table that does not fit in driver or executor memory causes OOM. In production, confirm BroadcastHashJoin in explain. Sort-merge is safer when both sides are large.

The first plan uses estimates. After a shuffle, Spark knows real sizes. That is when AQE rewrites. Enable with spark.sql.adaptive.enabled. AQE does not fix a bad scan, a Python UDF, or a huge collect. In production, still filter early and pick a sane join. Then let AQE clean leftover partition and skew issues.

Prove skew in Spark UI: max task time much larger than median, or one shuffle partition much larger than others. Fixes include AQE skew join, broadcast the small side, salting, isolating the hot key, or two-phase aggregation. Skew is a memory risk too, because one task may build a huge hash map.

If 80% of rows are country = US, one join partition does most of the shuffle. Salt turns US into US_0 ... US_9. That creates more partitions and more parallelism. The small side grows by the salt factor, so it must still fit. In production, try broadcast or AQE first. Salt when both sides are large and one key dominates.

DataFrames and Datasets use Tungsten. Plain RDD Python objects do not. Binary rows plus whole-stage codegen make CPU-bound SQL faster. Off-heap storage still uses RAM. It just avoids some JVM GC. In production, Python UDFs can leave Tungsten and slow the job.

Phases are analysis, logical optimization, physical planning, then code generation. Rules include predicate pushdown, column pruning, constant folding, and join selection. Catalyst cannot see inside a Python UDF, so filters after a UDF may not push down. In production, inspect the physical plan, not just the source code.

df.filter("year = 2024").select("id", "amount") can prune partitions and columns. A Python UDF filter often blocks pushdown because Spark cannot convert it to a source predicate. In production, filter and select early, use partitioned layouts, and keep types explicit so the predicate is valid.

It is a Tungsten feature applied after Catalyst picks a physical plan. It works well for filters, projections, and many aggregations. It may not apply across some joins, Python UDFs, or unsupported operators. In Spark UI, collapsed stages and *codegen* in the plan are clues. In production, prefer built-in functions so codegen can run.

If a stage has 400 partitions, you see about 400 tasks. Look at task time distribution. A long max vs median is skew. High shuffle spill or GC on a stage is a memory signal. The SQL tab shows the plan. In production, debug the slow stage first, then the slow tasks inside it.

Find the failed executor and stage in the UI and logs. Compare max task input and shuffle size with the median. Spark memory is split between execution and storage. Spill is a warning. It is not always a crash. Fixes include smaller partitions, salting, not broadcasting a big table, unpersisting cache, and avoiding groupByKey of huge lists.

Configs include spark.dynamicAllocation.enabled plus min, max, and idle timeout. Cached data on an executor can be lost if that executor is removed, so Spark may keep executors that hold cache. In production, set a max so one job cannot take the whole cluster. Speculative execution is a different feature.

Enable with spark.speculation. It wastes some CPU because two tasks do the same work. If the task is slow because of skew, the duplicate is also slow. Fix skew instead. In production, use speculation for noisy clusters. Use AQE or salting for hot keys.

Lineage is enough for short pipelines. Long iterative graphs and stateful streaming need a durable cut. rdd.checkpoint() is lazy until an action. A common pattern is cache then checkpoint so Spark does not compute twice. Checkpoint is slower than cache because it writes to HDFS or object storage.

Updates inside transformations can run more than once because Spark retries tasks. That double-counts. Prefer accumulators in actions such as foreach, or use DataFrame metrics instead. Accumulators are not a replacement for aggregations you need in the output table.

Without broadcast, Spark can ship a Python closure with every task. A 100 MB lookup times 1000 tasks is a lot of network. sc.broadcast(obj) then bc.value is the RDD pattern. DataFrame broadcast joins use the same idea. The object must fit in driver and executor memory.

df.createOrReplaceTempView("orders") then SQL is still a Spark job. There is no separate SQL cluster. Hive support lets Spark read metastore tables. In production, SQL and DataFrame quality is the same if the plan is the same. Check explain either way.

enableHiveSupport() and a metastore URI are the usual setup. Spark can create managed or external tables. Concurrent writers still need a table format or process that handles that, such as Hive ACID or Delta. In production, treat the metastore as a contract for schema and location.

spark.read.option("mergeSchema", "true").parquet(path) scans file footers to build a union schema. That listing cost hurts when you have many small files. unionByName(allowMissingColumns=True) is the DataFrame version when you already have two frames. In production, prefer a controlled schema and additive columns over silent merge.

PERMISSIVE is the usual default. Bad data can hide as nulls. FAILFAST is safer when a malformed file should page you. For production lakes, land raw files first, then quarantine bad rows into a dead-letter path. Do not silently drop keys you need for money or identity.

Configs include spark.sql.adaptive.coalescePartitions.enabled and advisoryPartitionSizeInBytes (often 64 MB). This is not the same as df.coalesce(). AQE runs after a shuffle stage. It cannot fix a scan that already created 50,000 tiny input files. In production, still compact writes. Let AQE clean shuffle leftovers.

This helps thin clients, multiple languages, and remote IDE workflows. The DataFrame API still looks similar. Some older RDD and context APIs may not map 1:1. In production, treat Connect as a client protocol, not a new execution engine.

Use it when a team knows pandas but data does not fit on one machine. For new ETL, the DataFrame API is usually clearer and closer to Spark features. to_pandas() still collects to the driver and can OOM. In production, check the Spark plan behind the pandas-like calls.

df.count() is one job. If the plan has two shuffles, you may see three stages. Stage 1 writes shuffle files. Stage 2 reads them. This map is how you talk in interviews and how you read Spark UI. Memory problems show up at the task that holds the fat partition.

Sort-merge is the default for large-large equi joins. It is robust but pays shuffle plus sort. AQE can switch to broadcast after a filter shrinks one side. Nested-loop joins are for non-equi conditions and can explode memory. Always verify the chosen strategy in explain.

groupByKey can build a huge in-memory list per key and OOM on skew. DataFrame groupBy is not the same as RDD groupByKey. Spark SQL aggregations can use partial maps. In production, never collect all events per user into one list unless the list is bounded.

Streaming micro-batches are a common cause. Fix by coalescing or repartitioning before write, compacting later, or using a table format optimize command. Aim for files around 128-512 MB depending on the engine. In production, watch output file count as a first-class metric.

A simple starting math is shuffle size divided by 128-200 MB. 10 GB of shuffle is tens of partitions, not 2000. spark.default.parallelism is more of an RDD default and is a different knob. After you set a sane baseline, let AQE coalesce. Measure stage time and spill, then change one config.

This is the first DataFrame question in many interviews. Use column expressions, not a Python if over collect(). A filter can be pushed into Parquet or partition folders, which cuts scan and later shuffle bytes. In production, filter before joins.

1. It starts from the employees DataFrame.
2. It keeps only rows with salary greater than 50000.
3. It selects a small set of columns.
4. It sorts by salary high to low.

Say the grain first: one row per customer. Alias the metrics so names are clear. Skewed customers can make one task huge. AQE or two-phase aggregation helps when one customer dominates.

1. It groups orders by customer_id.
2. It counts orders and sums amount in the same aggregation.
3. It sorts customers by total spend.

Window functions do not collapse rows. They add a rank column, then you filter. This shuffles by the partition key. dense_rank keeps ties. row_number would drop extras. In production, partition only by the group you rank on.

1. It sums spend per product in a category.
2. It ranks those products inside the category.
3. It keeps rank 1 and rank 2, including ties.

dense_rank treats the same salary as one level. row_number would pick an arbitrary second row even when salaries tie. A global window still shuffles to sort. For a huge table, approx_percentile is a different question.

1. It ranks every employee by salary.
2. Equal salaries share a rank.
3. It keeps rows whose rank is 2 and returns that salary.

rowsBetween(unboundedPreceding, currentRow) is the usual frame. Without a frame, some functions still default in a surprising way, so write the frame. A global running total needs a sort and can be heavy. In production, partition by account or region if the total should reset.

1. It orders revenue by date.
2. For each row it sums amount from the beginning through that row.
3. It stores that sum as running_total.

rowsBetween(-6, 0) is 7 rows, not 7 calendar days. If dates are missing, that is a different window. rangeBetween on a unix timestamp can mean real days. Rolling windows shuffle and sort. In production, partition by store or user.

1. It orders daily sales by date.
2. It averages sales over seven rows ending at the current date.
3. It rounds the rolling average to two decimals.

Alias both sides or column names collide. This join shuffles on the join key unless one side is tiny. Null manager_id is the CEO and should not match. In production, watch skewed manager ids if one person manages a huge org.

1. It joins each employee to the row of that employee's manager.
2. It keeps employees whose salary is higher than the manager salary.
3. It returns both names and both salaries.

A left join plus order_id IS NULL works, but left_anti is clearer and often planned well. left_semi is the opposite: left rows that do match. Anti joins still shuffle unless the right side is broadcast. Prefer anti join over NOT IN with nulls.

1. It looks for each customer in the orders table.
2. It keeps only customers with zero matching orders.
3. It returns those customer ids and names.

dropDuplicates() with no columns is DISTINCT on all columns. Dedup shuffles. For production CDC, "one row" must mean "the latest by event time." Then window + filter rn == 1 is the right pattern.

1. The first call keeps one row for each email value.
2. The second call keeps fully unique rows.
3. Neither call promises the newest timestamp.

lag(col, 1) reads the prior row in the window. Guard a zero previous revenue to avoid divide by zero. This window shuffles and sorts. Partition by product if MoM is per product, not global.

1. It orders revenue by year_month.
2. It copies last month's revenue onto the current row.
3. It computes percent change and turns a 0 previous into null.

Pivot scans distinct values unless you pass the list. That extra distinct is a shuffle. Too many distinct months or ids makes a very wide row and can OOM. Unpivot is the reverse, usually with stack or explode.

1. It keeps one row per product.
2. It turns each month value into a column.
3. It fills those columns with summed revenue.

Explode can multiply rows and shuffle more bytes in the next aggregate. Null arrays with explode remove the parent row. In production, cap array size before explode so one bad row cannot blow memory.

1. It turns each tags array into one row per tag.
2. It counts how often each tag appears.
3. It returns the five most used tags.

This is a gaps-and-islands pattern on time. lag gets the previous timestamp. Unix seconds make the 1800-second check easy. The cumulative sum window shuffles by user. Skewed bots with millions of events can OOM a single user partition.

1. It finds the previous event time for the same user.
2. It marks a new session when there is no previous event or the gap is over 30 minutes.
3. It sums those flags over time so each session gets an id.

This is the classic islands trick. Consecutive dates share the same date_sub(date, rn) key. The window and the groupBy both shuffle. Distinct first, or same-day duplicates fake a streak. In production, store dates as date, not messy timestamps.

1. It keeps one purchase date per customer.
2. It numbers those dates and builds an island key.
3. It keeps customers who have at least three days in one island.

This join can explode. An order with n items makes n*(n-1)/2 pairs. Cap items per order or you will blow shuffle and memory. In production, aggregate to distinct products per order first.

1. It joins items that share an order.
2. It keeps each unordered pair once by using product_id less than.
3. It counts pairs and returns the top five.

A user who views 10 times is still one viewer. max(when(...)) per user does that. groupBy event_type over-counts repeats. This is two aggregations and two shuffles. Collecting a 3-row summary to the driver is safe.

1. It builds one row per user with 0/1 flags for each stage.
2. It sums those flags to get unique users at each stage.
3. Conversion rates are purchased / viewed and similar ratios.

inferSchema on JSON is slow and unstable. Explicit schema is the production habit. Nested structs are still one column until you flatten. Deep explode of nested arrays can explode row count and memory.

1. It defines the nested JSON schema.
2. It parses the string column into a struct.
3. It lifts nested fields into flat columns.

This is the DataFrame form of schema merge. Position-based union silently mis-assigns columns when a new field appears in the middle. In production, still enforce a target schema after the union so types do not drift.

1. It reads two monthly folders that may differ in columns.
2. It stacks them by column name.
3. Columns that exist on only one side become null on the other.

A hot key such as US lands in one shuffle partition. Salt splits it into N keys. The small side grows by N, so it must still be manageable. This extra explode is memory and shuffle cost. In production, measure task times before and after.

1. It appends a random 0-9 suffix to each orders country key.
2. It copies each countries row once per suffix.
3. It joins on the salted key so the hot country spreads across tasks.

Auto-broadcast may not fire if stats are missing or the table is just over 10 MB. A hint fixes that. Broadcasting a 10 GB table OOMs the driver or executors. In production, this is the first join optimization for fact plus small dim.

1. It marks store_metadata as a broadcast side.
2. It joins transactions to stores on store_id without shuffling stores.
3. It prints the physical plan so you can see BroadcastHashJoin.

Partitioning helps predicate pushdown. Over-partitioning on a high-cardinality column such as user_id creates millions of tiny folders. Repartition by the same columns before write if you need fewer files per folder. Use overwrite with care so you do not wipe the whole table.

1. It derives a date column from the event timestamp.
2. It shuffles so each date tends to sit together.
3. It writes Parquet folders like event_date=2024-01-01.

withWatermark("event_time", "10 minutes") means data 10 minutes late can still join or aggregate. Later than that may be ignored. Watermarks need a streaming source and an event-time column. They are not a batch filter. In production, pick the delay from real late-arrival metrics.

1. It reads a streaming JSON source.
2. It sets a 10-minute late-data allowance on event_time.
3. It counts clicks in 5-minute event-time windows per campaign.

na.drop(subset=...) is row removal. na.fill is value replacement. Filling ids hides bad data. In production, send dropped rows to a quarantine path so you can measure data quality.

1. It removes rows with a null employee id.
2. It sets null salaries to 0.
3. It sets null departments to Unknown.

Built-ins stay in the JVM, can be pushed down, and use Tungsten. Python UDFs pay pickle and process hops. Pandas UDFs are better than row UDFs but still worse than native functions. In production, rewrite the UDF if Spark already has the function.

1. The UDF version computes length in Python per row.
2. The built-in version computes length inside Spark.
3. Both add a column n, but only the built-in stays fully optimizable.

Order matters. If you test >= 50000 before > 100000, high salaries become Mid. Null salary fails every comparison and falls into otherwise unless you handle isNull first. This is a narrow projection. No shuffle.

1. It labels null salaries as Missing.
2. It labels salaries above 100k as High.
3. It labels 50k to 100k as Mid and the rest as Low.

Explode is for when you need one row per element. Array functions avoid that row explosion and extra shuffle. Higher-order filter/transform run in the JVM. In production, bound array size. Huge arrays in one row still OOM a task.

1. It counts tags in the array.
2. It flags rows whose array includes spark.
3. It builds a new array of tags with length at most 10.

Top-N with ties usually wants dense_rank. Dedup "exactly one row" wants row_number. Competition ranking wants rank. The window shuffles by partition key. OrderBy must be deterministic or row_number is random among ties.

1. It computes three different ranks inside each department.
2. row_number gives a unique sequence.
3. rank and dense_rank show how ties are numbered.

dropDuplicates does not honor time. rank can keep two rows if timestamps tie. Add a tie breaker such as an ingest id. This window shuffles by the key. It is the standard CDC "current row" pattern.

1. It numbers events per user and event type, newest first.
2. It keeps only number 1.
3. It drops the helper column.

reduceByKey combines locally before shuffle. groupByKey does not. The DataFrame path uses Catalyst. Clean punctuation or you count Spark. and Spark as two words. In production this is a toy. The lesson is local aggregation.

1. It reads each line as a value column.
2. It splits and explodes words to lowercase.
3. It counts words and returns the top 10.

max in agg returns one metric row per dept, not the full employee row. People often want the person who has that salary. Then use row_number or join the max back. Aggregation shuffles on dept.

1. It groups employees by department.
2. It computes the highest salary in each group.
3. It does not return the employee name.

An inner join on orders would repeat the customer once per order. Semi join is the efficient "exists" form. Anti join is "not exists." Both still shuffle or broadcast the right side.

1. It checks whether each customer appears in orders.
2. It keeps customers who appear at least once.
3. It does not copy order columns or duplicate customers.

collect_list keeps duplicates and is unordered unless you use a window. collect_set dedups. This aggregation can hold a whole group in memory. In production, cap with slice or avoid collecting huge event histories.

1. It groups items by order.
2. It gathers product ids into an array.
3. It keeps at most 100 products so one fat order cannot explode memory.

date_format to 'yyyy-MM' also works but is a string. Truncate before grouping so the shuffle key is the month. In production, also decide timezone. UTC vs local can move late events to the next month.

1. It maps each order timestamp to the first instant of its month.
2. It sums amount per month.
3. It sorts months in time order.

Windows compare rows, not calendars, unless you check the date gap. This is a common LeetCode-style Spark question. A global orderBy shuffles the whole weather table.

1. It fetches yesterday's row using lag.
2. It checks that the previous row is exactly one calendar day earlier.
3. It keeps days hotter than that previous day.

Streak questions are islands questions. Distinct dates first. Then window plus groupBy. Skewed power users with years of daily logins still fit if you only store one row per day. Do not explode session events into the streak.

1. It keeps one login date per user.
2. It groups consecutive dates into islands.
3. It takes the longest island per user.

This is for a string column that happens to hold JSON, not for spark.read.json on files. Explicit schema avoids a second inference pass. In production, log null payload rates. That is your bad-record metric.

1. It declares the expected JSON fields.
2. It parses payload into a struct.
3. It selects the inner fields as columns.

Higher-order functions avoid a shuffle that explode plus groupBy would need. They still run per row, so a 10 million element array is a memory problem. Prefer them over Python UDFs that loop lists.

1. It reads an array column of prices.
2. It multiplies each element by 100.
3. It stores the new array as integer cents.

Use pandas UDFs for vector math Spark cannot express. You pay Arrow conversion and Python workers. They can increase executor memory. If F.cos exists, use it. In production, measure CPU and memory against a native rewrite.

1. It declares a pandas UDF that returns doubles.
2. It standardizes a batch of scores.
3. It adds a z column using Python vector code instead of a per-row UDF.

first without ignorenulls can return null. For the full first-order row, row_number ascending and descending is clearer. Two windows still shuffle once per partition key if they share the spec.

1. It groups orders by customer.
2. It takes the earliest order date.
3. It takes the latest order date.

Exact countDistinct shuffles a lot of unique values. Approx is the production default for dashboards. You can pass a relative standard deviation. Do not use approx for money or billing uniqueness.

1. It groups events by country.
2. It estimates unique users with HyperLogLog.
3. It also computes exact distinct for comparison.

This is Spark SQL grouping sets. It is more than one groupBy. Extra totals mean extra aggregation work and shuffle. In production, replace null totals with labels using grouping() so they are not confused with unknown country.

1. It aggregates revenue at country plus product.
2. It also emits country subtotals and a grand total.
3. It labels those extra rows using grouping_id.

Range joins are a common production surprise. Spark may pick BroadcastNestedLoopJoin. That is O(n*m) in the worst case. Bin events into time buckets and join on bucket plus a residual filter to make an equi join. Check the plan.

1. It broadcasts the small campaigns table.
2. It keeps events whose time falls in a campaign window.
3. It avoids shuffling campaigns.

This is a shuffle aggregation. An inner join back to users can duplicate the count onto every matching row. For huge tables, write the duplicate keys first, then join.

1. It counts rows per email.
2. It keeps emails that appear more than once.
3. It joins back to return the full duplicate rows.

This is a metadata column, not a shuffle. It helps quarantine. On some sources it can be empty, so test it. In production, store it in bronze tables.

1. It reads every matching Parquet path.
2. It records which file each row came from.
3. Later quality checks can group errors by source_file.

Object stores can hide deep date folders. Recursive listing has a metadata cost. Many tiny files will still hurt. Prefer Hive-style partitions when you can prune.

1. It walks nested folders under the events prefix.
2. It only loads files that match *.parquet.
3. It returns one DataFrame of those files.

Batch groupBy does not need a watermark. Streaming state does. Checkpoint location stores offsets and state. Without it, a restart duplicates or drops data. Output mode is usually append for windowed aggregations with watermarks.

1. It reads JSON as a stream and allows 15 minutes of lateness.
2. It counts events in 10-minute windows per country.
3. It writes Parquet and stores streaming state in a checkpoint.

explode drops the position. posexplode_outer keeps empty arrays. Positions start at 0. The row count still grows, so the same memory warning as explode applies.

1. It expands each products array.
2. It keeps the position of the item in the array.
3. It keeps the product id at that position.

Maps are not ordered. Do not treat keys as a stable schema. If keys are really columns, pivot after explode. Huge maps in one row are another OOM pattern.

1. It turns each map entry into a row with k and v.
2. It groups by the map key name.
3. It counts how often each property appears.

This is one of the most common name collisions in Spark interviews. F.coalesce is SQL COALESCE. df.coalesce(n) changes partition count. Mixing them up is an instant trap.

1. It looks at mobile first.
2. If mobile is null it uses home, then work.
3. It stores that value as best_phone.

Use ntile for even-sized groups. Use percent_rank or approx_percentile for value-based cuts. The window sort shuffles. Skewed values can put the same amount in different tiles.

1. It sorts orders by amount.
2. It splits that ordered list into four buckets.
3. It labels each row with quartile 1 to 4.

lag looks back. lead looks forward. Partition by customer or you will leak across users. Datediff then gives conversion delay. This is a shuffle sort per customer.

1. It orders each customer's purchases by date.
2. It copies the next purchase date onto the current row.
3. It computes the gap in days.

This is why windows exist: add group context without collapsing. partitionBy country gives country share. An empty partition spec gives global share. The window still shuffles by the partition key.

1. It computes total amount per country on every row.
2. It divides the row amount by that total.
3. It stores a percent of country sales.

This is a batch reconstruction of type-2 history. Real pipelines also handle late updates. The window shuffles by the business key. Overlapping ranges are a data bug. In production, prefer a merge into a Delta table, but the window logic is what interviews ask.

1. It finds the next start date for the same customer.
2. That next start becomes the current row's end date.
3. It flags the latest row as current.

This is the aggregation cousin of join salting. It helps when one country or one customer has most of the rows. AQE does not always fix groupBy skew. In production, try it when UI shows one reduce task holding almost all shuffle bytes.

1. It adds a random salt so one country splits across several partial groups.
2. It sums amount inside each salted group.
3. It sums those partials to the true country total.

Thousands of 2 MB files are a production incident. coalesce(1) is only for tiny results. Target file size around a few hundred MB. AQE coalescing shuffle partitions helps the compute stage. You still may need an explicit coalesce at write time.

1. It filters to one date, which can leave many almost empty partitions.
2. It merges those partitions down to 20.
3. It writes about 20 Parquet files instead of hundreds of tiny ones.