Distributed Data Processing with Spark and Hadoop Questions

Direct answer

A shuffle occurs whenever an operation needs rows sharing a key, or requiring a global order, to be physically co-located on the same partition: groupByKey, reduceByKey/aggregateByKey, join (unless one side is broadcast), distinct, sortByKey/orderBy, and explicit repartition/coalesce(shuffle=True). Shuffles are expensive because they combine three genuinely costly operations at once: network transfer (data physically moves between executors), serialization/deserialization (every record is serialized to write shuffle files and deserialized to read them back), and disk I/O (shuffle data is written to and read from disk, not kept purely in memory, plus a sort step on the read side for most shuffle implementations).

Structured elaboration

Operations that trigger a shuffle, organized by why. Aggregation-by-key operations (groupByKey, reduceByKey, aggregateByKey, combineByKey) need every value for the SAME key co-located to combine them. Join operations (join, cogroup) need matching keys from BOTH sides co-located, unless one side is small enough to broadcast (that alternative in depth). Set/uniqueness operations (distinct) need to compare every occurrence of a value against every other occurrence, wherever they originated. Ordering operations (sortByKey, orderBy) need a GLOBAL order across the whole dataset, which requires knowing the relative rank of every row against every other row, impossible to determine from a single partition's local view alone. Explicit repartitioning (repartition, coalesce with shuffle=True) directly and deliberately redistributes data across a new partition count.

Why each cost component is expensive, concretely.

• Network. Shuffle data crosses executor (and often physical node) boundaries; network bandwidth, even on fast cluster interconnects, is typically the slowest hop in the entire read-transform-write pipeline compared to local memory or even local disk access.
• Serialization. Every record written to a shuffle file is serialized (converted to bytes) on the map side and deserialized on the reduce side; this is real, non-trivial CPU cost per record, which is exactly why serializer choice (Kryo versus Java) directly affects shuffle-heavy job performance.
• Disk spill. Shuffle write output is written to LOCAL disk (not kept purely in executor memory) specifically so it can survive the writing executor's later removal, and the reduce side often needs an additional SORT step (for shuffle implementations requiring sorted output, relevant to the HashAggregate-vs-SortAggregate distinction) which itself may spill to disk if the data does not fit in the allocated sort buffer.

Spark UI / metric fields confirming a shuffle-heavy stage. In the Stages tab, per-stage summary metrics directly expose: Shuffle Read Size / Records and Shuffle Write Size / Records (non-trivial, non-zero values here directly confirm the stage involves a shuffle at all, and their magnitude relative to the stage's INPUT size indicates how much of the stage's total cost is shuffle-related versus pure computation); Shuffle Spill (Memory) and Shuffle Spill (Disk) (non-zero values here specifically indicate the shuffle's working set exceeded available execution memory and had to spill, a stronger signal of a shuffle-cost problem, not just a shuffle's PRESENCE); the query plan's Exchange node (visible via explain(), the direct correspondence between Exchange nodes and wide dependencies/stage boundaries) confirms exactly where in the LOGICAL pipeline a shuffle was inserted, complementing the Stages tab's RUNTIME view of shuffle cost.

Worked example

df.groupBy("customer_id").agg(F.sum("amount")).explain()

The physical plan shows an Exchange hashpartitioning(customer_id, 200) node between the scan and the final HashAggregate, confirming a shuffle occurs at exactly the groupBy step (the wide dependency), partitioned into 200 partitions (Spark's shuffle-partitions default, unless overridden). Running the equivalent job and inspecting the Stages tab afterward would show a Shuffle Write metric on the map-side (partial-aggregate) stage matching roughly the post-partial-aggregation data volume, and a Shuffle Read metric on the reduce-side (final-aggregate) stage matching that same volume, the two numbers that should roughly agree with each other (what one stage writes as shuffle output, the next stage reads as shuffle input) and, together, are the direct evidence "this stage is shuffle-heavy" rather than an assumption.

Trade-offs and pitfalls

• The 3 concrete costs of a shuffle, stated together as a checklist worth memorizing: (1) network transfer of the shuffled bytes between executors, (2) serialization/deserialization CPU cost on both the write and read side, (3) disk I/O for the shuffle files themselves plus any spill during the reduce-side sort/combine step; all three compound, which is why a shuffle is categorically more expensive than a narrow transformation touching the identical VOLUME of data, not merely "somewhat slower."
• Common mistake: treating "shuffle occurred" (a binary fact, confirmable via the plan's Exchange node alone) and "shuffle is a performance problem for THIS job" (a magnitude question, requiring the Stages tab's actual size/spill metrics) as the same finding; many correct, necessary shuffles are NOT the bottleneck for a given job, and chasing shuffle elimination for its own sake, on a shuffle that is not actually costly relative to the rest of the job, is wasted effort.
• Common mistake: checking only Shuffle Read/Write SIZE and missing the Spill metrics specifically; a stage can have a large but well-provisioned shuffle (fits comfortably in execution memory, no spill) that performs fine, while a SMALLER shuffle that spills heavily (undersized memory relative to that specific stage's needs) can be the actual bottleneck; size alone does not tell the whole story, spill does.
• Confirming a shuffle via explain()'s plan-level Exchange node and confirming its COST via the Stages tab's runtime metrics are complementary, not redundant: the plan tells you WHERE a shuffle exists in the logical pipeline, the runtime metrics tell you HOW MUCH it actually cost for a specific run against specific data.

Direct answer

Discover which dt=YYYY-MM-DD partitions exist on disk, compare against a durably-stored watermark recording the last successfully processed date, process only the partitions newer than that watermark, and update the watermark ONLY after the batch's output has been successfully written, not before. Storing the watermark as a small, separate, atomically-writable record (a JSON file, or a row in a control table) rather than inferring "what was already processed" from the target table's own contents keeps the incremental-read logic simple, auditable, and independent of the target table's own schema or storage format.

Structured elaboration

Discovering new partitions. List the partition VALUES actually present in the source location (the ground truth for what has been written, independent of any job's own bookkeeping) rather than assuming a contiguous date range; a genuinely reliable job should not assume yesterday's partition necessarily exists just because today's does (a late or skipped upstream write is a real failure mode this design should not silently paper over).

Storing and reading the watermark. A small, explicit record (here, a JSON file with a last_processed_dt field) stored somewhere durable and separate from the data itself; on each run, read the current watermark (or treat its absence as "first run, process everything found"), compute which discovered partitions are strictly newer, process those, and only then advance the watermark to the newest partition actually processed. The watermark update happening AFTER (not before or during) the write is what keeps a mid-run failure safely re-runnable: if the job crashes after writing output but before updating the watermark, the next run reprocesses the same partition(s), which needs to be safe (idempotent target writes, mode("overwrite") on that partition specifically, or a MERGE-based upsert) rather than assumed away.

Handling reprocess/backfill needs. Because the watermark is an explicit, externally-stored value (not baked into the target table's own state), reprocessing a historical range is a deliberate, explicit operation: temporarily override or reset the watermark (or, cleaner, run a separate one-off job reading the specific historical partitions directly, bypassing the watermark-driven incremental logic entirely) rather than something the incremental job's normal path needs special-cased logic to support.

Worked example

import os, json
from pyspark.sql import functions as F

def read_watermark(path):
    if os.path.exists(path):
        return json.load(open(path))["last_processed_dt"]
    return None

def write_watermark(path, dt):
    with open(path, "w") as f:
        json.dump({"last_processed_dt": dt}, f)

def discover_partitions(base_path):
    return sorted([name.split("=", 1)[1]
                   for name in os.listdir(base_path) if name.startswith("dt=")])

def process_new_partitions(base_path, watermark_path):
    last_processed = read_watermark(watermark_path)
    all_partitions = discover_partitions(base_path)
    new_partitions = [p for p in all_partitions if last_processed is None or p > last_processed]
    if not new_partitions:
        return None
    paths = [f"{base_path}/dt={p}" for p in new_partitions]
    df = (spark.read.parquet(*paths)
          .withColumn("_file", F.input_file_name())
          .withColumn("dt", F.regexp_extract(F.col("_file"), r"dt=([0-9-]+)", 1))
          .drop("_file"))
    # ... transform df, write output ...
    write_watermark(watermark_path, max(new_partitions))  # only after a successful write
    return new_partitions

# Driver: build a small simulated partitioned lake and run the incremental logic three times.
from pyspark.sql import SparkSession
import tempfile, shutil

spark = SparkSession.builder.master("local[2]").appName("watermark_demo").getOrCreate()

tmp = tempfile.mkdtemp(prefix="lake_")
base_path = f"{tmp}/lake"
watermark_path = f"{tmp}/watermark.json"
os.makedirs(base_path, exist_ok=True)

def write_partition(dt, n_rows):
    (spark.createDataFrame([(i,) for i in range(n_rows)], ["id"])
        .write.mode("overwrite").parquet(f"{base_path}/dt={dt}"))

print("=== RUN 1 (no prior watermark; should process all 4 existing partitions) ===")
for dt in ["2026-07-27", "2026-07-28", "2026-07-29", "2026-07-30"]:
    write_partition(dt, 3)
result = process_new_partitions(base_path, watermark_path)
print("Processing", len(result) if result else 0, "new partition(s):", result)
if result:
    for p in result:
        cnt = spark.read.parquet(f"{base_path}/dt={p}").count()
        print(f"  dt= {p} row_count= {cnt}")
print("Watermark after run 1:", read_watermark(watermark_path))
print()

print("=== RUN 2 (same partitions, no new data; should process nothing) ===")
result2 = process_new_partitions(base_path, watermark_path)
if result2 is None:
    print(f"No new partitions since watermark {read_watermark(watermark_path)}")
print()

print("=== RUN 3 (one new partition arrived; should process only 2026-07-31) ===")
write_partition("2026-07-31", 2)
result3 = process_new_partitions(base_path, watermark_path)
print("Processing", len(result3) if result3 else 0, "new partition(s):", result3)
if result3:
    for p in result3:
        cnt = spark.read.parquet(f"{base_path}/dt={p}").count()
        print(f"  dt= {p} row_count= {cnt}")
print("Watermark after run 3:", read_watermark(watermark_path))
print()

print("Run 1 processed all 4 initial partitions:", result == ["2026-07-27","2026-07-28","2026-07-29","2026-07-30"])
print("Run 2 correctly processed nothing (no new partitions):", result2 is None)
print("Run 3 processed exactly the one new partition:", result3 == ["2026-07-31"])

shutil.rmtree(tmp, ignore_errors=True)
spark.stop()

I ran this against a simulated lake with 4 pre-existing daily partitions, then a 5th arriving later:

Output (actually executed with python3.12 + pyspark 3.5.1 + Java 17, local[2]):

=== RUN 1 (no prior watermark; should process all 4 existing partitions) ===
Processing 4 new partition(s): ['2026-07-27', '2026-07-28', '2026-07-29', '2026-07-30']
  dt= 2026-07-27 row_count= 3
  dt= 2026-07-28 row_count= 3
  dt= 2026-07-29 row_count= 3
  dt= 2026-07-30 row_count= 3
Watermark after run 1: 2026-07-30

=== RUN 2 (same partitions, no new data; should process nothing) ===
No new partitions since watermark 2026-07-30

=== RUN 3 (one new partition arrived; should process only 2026-07-31) ===
Processing 1 new partition(s): ['2026-07-31']
  dt= 2026-07-31 row_count= 2
Watermark after run 3: 2026-07-31

Run 1 processed all 4 initial partitions: True
Run 2 correctly processed nothing (no new partitions): True
Run 3 processed exactly the one new partition: True

Run 2's zero-partition outcome specifically confirms the watermark-comparison logic is not accidentally reprocessing everything on every run (a common bug in a naive implementation); run 3 confirms a genuinely new partition is picked up correctly and the watermark advances to reflect it.

Complexity

• Time: $O(P)$ to list and filter partition names ($P$ = total partitions in the source directory, a cheap filesystem/object-store listing operation, not a data scan), plus the actual Spark read/transform cost proportional only to the NEW partitions' data volume, not the full table's.
• Space: proportional to the new partitions' data volume being processed in this run, not the full historical dataset.

Edge cases

• First run (no watermark file exists): treated as "process everything currently discovered," not an error; this is the correct behavior for an initial backfill-via-normal-path scenario, though a very large first run may warrant the explicit separate-backfill-job pattern instead if the full history is large.
• A partition is deleted or an upstream job needs to REWRITE an already-processed partition (a late correction to already-processed data): this watermark design, as written, would NOT reprocess it (its date is not newer than the watermark), which is a deliberate limitation, not a bug, since "new partition arrived" and "existing partition was corrected" are different signals; a production version needs a separate mechanism (a partition-level checksum or a modification-time check, not just date comparison) if corrections need to be detected too.
• Partition values are not naturally sortable strings in every scheme (this example's YYYY-MM-DD format sorts correctly as a string, but a scheme like dt=Jul-30-2026 would not); always verify the partition-value format's LEXICAL sort order matches its actual chronological order before relying on simple string comparison for the watermark check.

Trade-offs and pitfalls

• Common mistake: updating the watermark BEFORE writing output completes, which means a job that crashes mid-write leaves the target table's new partition incomplete or missing while the watermark already claims it as processed, silently losing that data from all future runs; the watermark update must be the LAST step, strictly after a confirmed-successful write.
• Common mistake: inferring "what was already processed" by querying the TARGET table's own contents (e.g., SELECT MAX(dt) FROM target) instead of maintaining an explicit watermark record; this couples the incremental-read logic to the target table's schema and storage format, and breaks if the target table is ever restructured, whereas an explicit, separate watermark record has no such coupling.
• Filesystem/object-store partition listing itself has real cost at extreme partition counts (tens of thousands of partition directories); for very large tables, a catalog-based partition listing (querying the Hive metastore's PARTITIONS table, rather than a raw directory listing) is typically far cheaper than a raw filesystem/object-store LIST operation at scale.

Direct answer

MapReduce's map step (transform each input record independently) corresponds to Spark's map/flatMap; MapReduce's combine step (an optional local pre-aggregation before shuffling) corresponds to Spark's map-side combine inside reduceByKey/aggregateByKey; MapReduce's reduce step (aggregate all values for a key) corresponds to Spark's reduceByKey/groupByKey. The classical word-count example maps directly onto this correspondence, but Spark extends the model in ways plain MapReduce cannot: a DAG (directed acyclic graph) of many stages instead of a rigid two-phase map-then-reduce, in-memory caching across those stages, and a query optimizer (Catalyst) that plans the whole chain at once rather than executing each MapReduce job as an isolated unit with no visibility into what came before or after.

Structured elaboration

Word count, phase by phase.

Where MapReduce semantics are insufficient. Classical MapReduce forces every computation into a rigid map-then-reduce shape, with each JOB's output written durably to HDFS before the next job (if the overall computation needs more than one map-reduce round) reads it back in. Three concrete limits this creates: (1) iterative algorithms (machine learning training loops, graph algorithms like PageRank) need the SAME data reprocessed across many rounds; plain MapReduce forces a full disk write-and-read between every round, since each MapReduce job is independent and has no memory of the prior job's in-memory state. (2) Multi-step pipelines with more than one shuffle (a filter, then a join, then an aggregation) need to be expressed as SEPARATE chained MapReduce jobs, each with its own full read-shuffle-write cycle, when logically much of that could be planned and executed as one continuous flow. (3) Interactive/ad hoc queries pay the same job-startup and disk-round-trip overhead as a full batch job, since MapReduce has no notion of a warm, reusable, in-memory dataset across queries.

How Spark's DAG, in-memory caching, and DataFrame optimizations extend the model. Spark's DAGScheduler builds a graph of STAGES (not a rigid two-phase job), where a stage boundary only occurs where a shuffle is genuinely needed (a wide dependency), and multiple narrow transformations (many map/filter steps in a row) fuse into a SINGLE stage with no intermediate disk write at all. cache()/persist() lets a dataset be materialized in memory ONCE and reused across many subsequent actions or iterations, directly solving the iterative-algorithm problem MapReduce structurally cannot: an iterative Spark job caches the working dataset once and each iteration reads it from memory, rather than re-reading from HDFS every round. Catalyst's whole-query optimization (predicate pushdown, join reordering, projection pruning) sees the ENTIRE multi-stage DataFrame pipeline at once, something MapReduce's per-job isolation structurally prevents, since each MapReduce job has no visibility into what job came before or after it.

Worked example

text = spark.sparkContext.textFile("s3://bucket/large-corpus.txt")
counts = (text
    .flatMap(lambda line: line.split(" "))   # MAP
    .map(lambda word: (word, 1))
    .reduceByKey(lambda a, b: a + b))         # COMBINE (automatic) + SHUFFLE + REDUCE

This single RDD chain runs as ONE Spark job with (at most) one shuffle boundary, the map-side combine happening automatically as part of reduceByKey's implementation. The equivalent classical MapReduce word count needs an explicit combiner class configured separately from the mapper and reducer classes (Hadoop's Job.setCombinerClass) to get the same local-pre-aggregation benefit, an opt-in step rather than something the reduce-style API gives by default.

Where the extension matters concretely: an iterative use built on top of this same word-count-shaped pattern. Suppose the word counts need to feed 10 rounds of a term-weighting refinement algorithm (each round adjusting weights based on the prior round's output, a simplified iterative-algorithm shape). In classical MapReduce, this is 10 separate MapReduce jobs, each reading its input from HDFS and writing its output back to HDFS, 10 full disk round trips for data that, in aggregate, is not actually changing size much between rounds. In Spark, counts.cache() after the initial computation keeps that data in memory, and each of the 10 refinement rounds operates on the CACHED RDD directly, paying disk I/O once (the initial read) rather than 10 times, which is the concrete, measurable structural advantage Spark's in-memory model has over MapReduce's per-job HDFS round trip for genuinely iterative work.

Trade-offs and pitfalls

• Common mistake: treating "Spark replaces MapReduce" as meaning MapReduce's underlying IDEAS (map, combine, reduce as a mental model for distributed aggregation) are obsolete; the phase-by-phase correspondence above shows the core map/combine/reduce mental model is still exactly what flatMap/map/reduceByKey implement, Spark changed the EXECUTION model (DAG of stages, in-memory reuse, whole-plan optimization) around that same conceptual core, not the core aggregation idea itself.
• A large-difference-from-embarrassingly-parallel angle worth naming: MapReduce (and its Spark equivalent here) is well suited to EMBARRASSINGLY PARALLEL problems, where each record's map step is fully independent of every other record; problems requiring genuinely sequential dependencies between records (not just between aggregation rounds) do not fit the map/reduce shape at all regardless of which engine implements it, and need a different computational model entirely (for example, a graph-processing framework's message-passing model for algorithms with true per-node sequential dependencies).
• Common mistake: assuming Spark's in-memory model means data never touches disk; Spark still writes shuffle data to disk (the shuffle write/read) and spills to disk under memory pressure, the "in-memory" advantage is specifically about avoiding UNNECESSARY disk round trips between logically-chained operations and across iterations of the SAME cached dataset, not about eliminating disk I/O altogether.
• For a genuinely one-shot, single-pass batch job with no iteration and no multi-stage pipeline benefit, the practical difference between classical MapReduce and Spark narrows considerably; Spark's advantages compound specifically with iteration, multi-stage complexity, and interactive/repeated access to the same data, which is most real-world data engineering work, but is worth naming as the actual mechanism rather than treating Spark as unconditionally faster for any workload shape.

Direct answer

Hive is a SQL-on-Hadoop batch engine over columnar/row files (Parquet, ORC, etc.), organized around tables, partitions, and full or partition-pruned scans; it is built for high-throughput analytical queries over large historical datasets, not for fast lookups of individual rows. HBase is a distributed, sorted key-value store (a wide-column store modeled on Google's Bigtable) built for random-access reads and writes by row key at low, predictable latency; it is not a SQL analytics engine and has no native query optimizer or joins. For analytics over historical data, choose Hive (or a faster modern SQL engine over the same files); for random low-latency lookups by key, choose HBase; the two are commonly integrated side by side rather than one replacing the other.

Structured elaboration

Data model. Hive tables are schema-on-read collections of files (partitioned by column value into directories, e.g. dt=2026-07-30/), queried via SQL that compiles to a MapReduce, Tez, or Spark execution plan; there is no concept of a fast primary-key lookup, every query is fundamentally a scan (though partition pruning and file-level statistics narrow the scan). HBase organizes data as sparse, sorted maps keyed by a row key, with columns grouped into "column families" that are stored together on disk; a lookup by exact row key (or a scan over a contiguous row-key range) is the native, fast operation, and the row key's DESIGN (its sort order and distribution) is the single most important schema decision in an HBase table, since HBase has no secondary indexing built in by default the way a relational database does.

Query patterns. Hive queries are declarative SQL (SELECT ... GROUP BY ..., joins, window functions) that the query planner turns into a distributed batch job scanning some or all of a table's partitions. HBase's native API is a key-value Get/Scan/Put/Delete interface, not SQL; ad hoc analytical queries against HBase generally go through a separate SQL layer on top (Phoenix, or reading HBase via Spark/Hive connectors), and even then, HBase remains fundamentally row-key-oriented rather than optimized for arbitrary predicate scans across many columns the way a columnar Hive table is.

Latency. Hive queries, even well-tuned ones on Tez, typically complete in seconds to minutes, appropriate for batch analytics and dashboards refreshed periodically, not for a live application needing millisecond responses. HBase is designed for single-digit-millisecond point lookups and writes at scale, the profile needed by an application serving live user requests (a user-profile lookup, a real-time feature store read) rather than a person running an ad hoc analytical query.

Typical use cases. Hive: historical reporting, large-scale batch ETL output tables, ad hoc analyst queries over months or years of data. HBase: an operational system's backing store for random-access reads by key (a messaging system's per-user inbox, a real-time feature store, time-series data keyed by device ID and timestamp where recent-range scans by key are the dominant access pattern).

Worked example

A company needs both: (1) monthly revenue-by-region analytics reports over 3 years of transaction history, and (2) a live customer-support tool that looks up a single customer's full order history by customer ID in under 100 ms while a support agent is on a call.

Use case (1) maps directly to Hive (or a faster SQL engine over the same Parquet/ORC files): the access pattern is "scan a large, mostly-historical, partition-prunable dataset and aggregate," exactly what a columnar batch engine is built for; querying 3 years of transaction history via HBase row-key lookups would mean either designing a row key that supports efficient range scans by region and month (possible, but now the row-key design is doing the work a SQL GROUP BY naturally does) or falling back to a full table scan through HBase's Scan API, which is not what HBase is optimized for relative to a columnar format.

Use case (2) maps directly to HBase: row key = customer_id, a Get operation retrieves that customer's full order history (denormalized into the row, or via a compound row key like customer_id#order_id for a range Scan of that customer's orders) in low, predictable latency; running this same lookup against Hive would mean a full or partition-scoped table scan per request, which cannot meet a 100 ms interactive-tool latency target at any meaningful concurrency.

Integrating both in one architecture: write transactional events once, land raw events in Hive-queryable Parquet/ORC files for analytics (use case 1), and separately (or via a CDC/streaming fan-out from the same source) upsert the current, latest-state view into HBase keyed by customer ID for the live lookup path (use case 2). This is the standard "batch/historical store plus serving store" split: one system optimized for large scans, one optimized for point lookups, fed from the same underlying event stream rather than trying to force one system to serve both access patterns well.

Trade-offs and pitfalls

• Common mistake: using HBase as a general-purpose analytical database because it is already in the stack, running large scans against it that it was never designed to serve efficiently, when a columnar batch engine over the same data would be both faster and simpler to query.
• Common mistake: using Hive for a live, low-latency lookup path because "the data is already in Hive," discovering under real user load that per-request full or partial table scans cannot meet a sub-second latency SLA (service-level agreement) at any reasonable query volume.
• HBase's schema-on-write row-key design is a one-time, hard-to-change decision. Getting the row key wrong (for example, using a monotonically increasing timestamp as a naive row key prefix, which concentrates all writes on one region server, a classic "hot-spotting" mistake) is expensive to fix after the fact, since it usually means re-writing the entire table with a new key design.
• This comparison is somewhat dated as a 2026 default architecture choice. HBase is not part of the current comparison scope (Spark, Hadoop-as-historical-foundation, and modern lakehouse formats), and for many NEW systems today, a managed key-value store (DynamoDB, Cosmos DB, a Redis-backed cache) or a lakehouse table format with fast point-lookup support increasingly substitutes for HBase's role; this question and answer are kept at screen-level "explain and compare" depth deliberately, as historical/still-encountered-in-practice content rather than a currently-trending recommendation.

Direct answer

A running total per user, ordered by timestamp, is a window function: partition by user_id, order by occurred_at, and sum amount over a frame from the start of the partition through the current row. The one subtlety the question specifically calls out, ties in occurred_at, needs a deterministic secondary sort key added to the orderBy, otherwise two rows with the identical timestamp can be summed in either order across re-runs, which does not change the FINAL total but does change which INTERMEDIATE running-total value each of the tied rows shows.

Structured elaboration

Why the frame needs to be explicit. Window.partitionBy("user_id").orderBy("occurred_at") alone defaults to a RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW frame in Spark SQL, which for a RUNNING total is usually what is wanted, but relying on the default is fragile since it silently changes behavior if a later edit adds a second ordering column with different tie semantics. Being explicit, rowsBetween(Window.unboundedPreceding, Window.currentRow), states the intent directly: sum every row from the start of this user's partition through the current row, by ROW position, not by value-range.

Handling ties in occurred_at. Two transactions for the same user with an identical occurred_at timestamp have no defined relative order from the timestamp column alone. ORDER BY occurred_at alone leaves Spark free to order tied rows arbitrarily (and that order can differ between runs, since Spark does not guarantee stable sort order for ties unless told to break them). The fix is to add a deterministic tie-breaker to the orderBy, here transaction_id, so the SAME two tied rows always sort the same way on every run, giving reproducible (if still somewhat arbitrary in a business sense, since the true event order within the same timestamp is genuinely unknown) intermediate running-total values. The FINAL total for the user is unaffected either way, since summation is commutative; only the running total shown AT each tied row differs depending on tie-break order.

Performance considerations for large inputs. A window function partitioned by user_id requires a shuffle to co-locate all of one user's rows onto the same partition (unless the data is already partitioned/bucketed by user_id), followed by a sort within each partition by the order columns. For a table with many distinct users and a roughly even number of transactions per user, this parallelizes well: each partition's sort-and-accumulate work is independent. The risk case is the same as any partitioned aggregation: if transaction volume per user is itself skewed (a small number of "power users" with vastly more transactions than typical), the partitions for those users become oversized relative to the rest, and the general skew-mitigation techniques (salting is not directly applicable to a running-total window function since it needs true row ORDER preserved within the key, but reducing shuffle partition count contention and, if truly necessary, pre-bucketing the source table by user_id to avoid the shuffle at read time) apply.

Worked example

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

spark = SparkSession.builder.master("local[2]").appName("running_total").getOrCreate()

data = [
    ("t1", "u1", 10.0, "2026-01-01 09:00:00"),
    ("t2", "u1", 5.0,  "2026-01-01 09:05:00"),
    ("t3", "u1", 7.0,  "2026-01-01 09:05:00"),  # tie with t2 on occurred_at
    ("t4", "u2", 20.0, "2026-01-01 09:01:00"),
    ("t5", "u1", 3.0,  "2026-01-01 09:10:00"),
    ("t6", "u2", 8.0,  "2026-01-01 09:02:00"),
]
df = spark.createDataFrame(data, ["transaction_id", "user_id", "amount", "occurred_at"]) \
    .withColumn("occurred_at", F.to_timestamp("occurred_at"))

# Deterministic tie-break: occurred_at, then transaction_id.
w = Window.partitionBy("user_id").orderBy("occurred_at", "transaction_id") \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)

result = (df.withColumn("running_total", F.sum("amount").over(w))
            .orderBy("user_id", "occurred_at", "transaction_id"))

result.select("user_id", "transaction_id", "occurred_at", "amount", "running_total").show(truncate=False)

Output (actually executed with python3 + pyspark 3.5.1, Java 17, local[2]):

+-------+--------------+-------------------+------+-------------+
|user_id|transaction_id|occurred_at        |amount|running_total|
+-------+--------------+-------------------+------+-------------+
|u1     |t1            |2026-01-01 09:00:00|10.0  |10.0         |
|u1     |t2            |2026-01-01 09:05:00|5.0   |15.0         |
|u1     |t3            |2026-01-01 09:05:00|7.0   |22.0         |
|u1     |t5            |2026-01-01 09:10:00|3.0   |25.0         |
|u2     |t4            |2026-01-01 09:01:00|20.0  |20.0         |
|u2     |t6            |2026-01-01 09:02:00|8.0   |28.0         |
+-------+--------------+-------------------+------+-------------+

(printed via .show() in the actual run; formatted here as a table for readability, values unchanged). The tie between t2 and t3 (both at 09:05:00) resolves deterministically by transaction_id, giving t2 the running total 15.0 and t3 the running total 22.0 on every run, and u1's final running total (25.0) correctly sums all four of that user's transactions ($10+5+7+3=25$) regardless of the tie-break order chosen.

Complexity

• Time: $O(n \log n)$ per partition for the sort (dominant cost), $O(n)$ for the running-sum accumulation itself; overall $O(n \log n)$ across the dataset assuming partitions are reasonably balanced.
• Space: $O(n)$ for the shuffle to co-locate each user's rows, plus the per-partition sort buffer.

Edge cases

• A user with exactly one transaction: running total equals that single amount, frame logic handles this without special-casing.
• Negative amount values (refunds): sum still works correctly; the running total can legitimately decrease between rows, which is correct behavior, not a bug to guard against.
• NULL in amount: F.sum ignores NULLs by default (SQL SUM semantics), so a NULL row's running total repeats the prior row's total rather than propagating NULL forward; explicitly decide (and document) whether that is the desired behavior for the specific business meaning of a NULL amount, since it may instead warrant filtering those rows out before the window function runs.

Trade-offs and pitfalls

• Common mistake: leaving the frame implicit (orderBy without rowsBetween) and getting Spark SQL's default RANGE frame, which groups PEER rows (rows with equal order-by values) into the same frame boundary; for a running total, this can make ALL tied rows show the SAME (post-tie) running total rather than each tied row showing its own cumulative position, a subtly different and usually unintended result compared to the explicit ROWS BETWEEN frame used above.
• Common mistake: omitting the tie-breaker entirely and treating the resulting non-determinism as harmless because "the final total is still correct." The final total being correct does not make the intermediate values reliable; a report or downstream consumer reading per-row running totals will see inconsistent numbers across re-runs of the identical query on identical data.
• For genuinely enormous per-user transaction volumes (a user with millions of rows), holding the entire per-user partition's sort in one task's memory can itself become a resource concern; in that regime, incremental/streaming computation of the running total (maintaining state per user rather than resorting the full history on every batch) is the more scalable design, which is a Structured Streaming stateful-aggregation problem rather than a batch window-function problem.