Data systems engineering, part 2: turning a SQL transformation into a Spark job

The first project in this series used DuckDB to turn ecommerce CSV files into trusted revenue. The query was small enough to fit in one local database, but its questions were already architectural:

• What is the grain of the data?
• Which rows are eligible for a business metric?
• What does a repeated execution do?
• What evidence proves that the result is correct?

The next step was to run the same kind of workload through Apache Spark. The goal was not to make twelve rows faster. The goal was to expose what changes when a relational transformation becomes a distributed DataFrame computation.

The companion learning repository contains the runnable application and tests. This article records the mental model, the observed output, and the physical plan that we can revisit before working on larger Spark workloads.

The workload: reproduce a known business result

The source is the clean ecommerce orders file from the first POC. It contains one row per order in this deliberately small model:

The business query is:

Keep completed orders.
Group them by order_date.
Sum amount.
Return the dates in order.

The important semantic detail is that amount is already the extended order amount. It must not be multiplied by quantity a second time.

The DuckDB baseline was:

Spark is not allowed to produce a merely plausible answer. It has to reproduce this known result while showing how the work is planned and executed.

The local tool boundary

The experiment used this local stack:

uv          manages the project environment
Python      launches the PySpark application
PySpark     provides the Python API and packaged Spark runtime
Java 17     runs Spark's JVM engine
Spark       plans and executes DataFrame work
CSV         supplies the source records
Parquet     stores the later physical-layout experiment
pytest      checks the invariants

The application uses:

SparkSession.builder \
    .master("local[2]") \
    .appName("poc2-inspect-orders") \
    .getOrCreate()

local[2] means that the Spark application runs in one local process with two local execution threads. It is useful for learning the execution model, but it is not equivalent to a multi-node Spark cluster. There are no remote machines in this experiment and no claim that twelve rows constitute a production benchmark.

The reproducible command is run from the 02-spark-batch directory of the learning repository:

bash scripts/run_poc2.sh --daily-revenue

To keep the Spark application alive for inspection:

bash scripts/run_poc2.sh \
  --daily-revenue \
  --enable-ui \
  --keep-alive

The wrapper selects the project Python interpreter, PySpark package directory, and Java 17. It avoids treating a missing standalone spark-submit command as proof that the project cannot reach Spark through its packaged runtime.

Explicit schema before transformation

The reader supplies an explicit schema rather than asking Spark to infer types:

ORDERS_SCHEMA = T.StructType(
    [
        T.StructField("order_id", T.StringType(), nullable=True),
        T.StructField("customer_id", T.StringType(), nullable=True),
        T.StructField("product_id", T.StringType(), nullable=True),
        T.StructField("order_date", T.DateType(), nullable=True),
        T.StructField("quantity", T.LongType(), nullable=True),
        T.StructField("amount", T.DecimalType(18, 2), nullable=True),
        T.StructField("status", T.StringType(), nullable=True),
    ]
)

The observed input boundary was:

records_read=12
input_partitions=1

The resulting schema was:

order_id:    string
customer_id: string
product_id:  string
order_date:  date
quantity:    long
amount:      decimal(18,2)
status:      string

The explicit schema is more than a convenience. A type inference decision is a contract decision. A date accidentally read as a string, or a monetary value read as a floating-point number, can change filtering, grouping, comparison, or financial accuracy.

A DataFrame is a plan before it is work

A useful first approximation is:

DataFrame transformation:
  Describe what should happen.

Action:
  Ask Spark to produce an observable result.

The following operations build a transformation:

completed = orders.filter(F.col("status") == "completed")
by_date = completed.groupBy("order_date").agg(F.sum("amount"))
ordered = by_date.orderBy("order_date")

The calls describe a logical computation. They do not mean that all twelve rows have already passed through every step. An action such as count(), show(), collect(), or a write causes Spark to plan and execute the work.

That distinction is called lazy evaluation. It lets Spark inspect the entire lineage before choosing a physical strategy. It can push a filter toward the source, remove unused columns, choose a join algorithm, and insert exchanges where data must be redistributed.

This is similar to a compiler or build system: constructing a graph is different from executing the compiled work. The analogy stops at the semantics. Spark's nodes process records and partitions, and failures can leave distributed state, shuffle files, or partial output that a build graph does not have to model in the same way.

Transformations and actions in the observed program

The application performs these observable operations:

Read CSV with explicit schema       transformation definition
Filter completed orders             transformation
Group by order_date                 transformation
Sum amount                          transformation
Order the result                   transformation

count()                             action
show()                              action

The program prints the physical plan before displaying the daily result. The result matched the DuckDB baseline:

+----------+-------------+
|order_date|daily_revenue|
+----------+-------------+
|2026-03-01|1350.00      |
|2026-03-02|1250.00      |
|2026-03-03|1725.00      |
|2026-03-04|2100.00      |
|2026-03-05|1200.00      |
|2026-03-06|2350.00      |
+----------+-------------+

Reading the physical plan from the inside out

The observed plan was abbreviated by Spark as:

AdaptiveSparkPlan
+- Sort
   +- Exchange
      +- HashAggregate
         +- Exchange
            +- HashAggregate
               +- Project
                  +- Filter
                     +- Scan csv

Read it from the bottom upward:

1. Scan csv reads the source file.
2. Filter keeps completed orders.
3. Project keeps the fields needed by the aggregation.
4. The first HashAggregate performs a partial sum locally.
5. The first Exchange redistributes rows by order_date.
6. The second HashAggregate combines partial sums for each date.
7. The second Exchange arranges the final result for the requested ordering.
8. Sort produces the dates in order.

The exact plan is evidence of a physical strategy, not merely a prettier version of the source code.

The filter was pushed toward the source

The scan reported:

PushedFilters: [IsNotNull(status), EqualTo(status,completed)]

Spark recognized that the status predicate could be pushed toward the CSV scan. This does not mean that a CSV file has gained Parquet's full metadata machinery. It means Spark is trying to avoid carrying rows that cannot satisfy the predicate further through the plan.

The first exchange is caused by aggregation

Before the grouping, rows can be spread across different Spark partitions. To calculate one complete sum for a date, all rows for that date need to meet at one logical aggregation partition.

The plan shows:

Exchange
Arguments: hashpartitioning(order_date, 200)

This is a wide dependency. Records cross a partition boundary, so Spark must serialize and redistribute data before the final aggregation can be complete.

The plan also shows a partial aggregate before the exchange. Spark first sums what it can locally, then moves smaller partial results rather than moving every raw row all the way to the final aggregation. That is a common distributed optimization.

The second exchange is caused by ordering

The final orderBy("order_date") requires globally ordered output. A local sort inside one partition is not enough to order data across all partitions. Spark therefore introduces another exchange, this time using range partitioning, then sorts the result.

This explains why a short SQL expression can create multiple physical stages. The business statement is “sum by date and order the result.” The engine must turn that into local work, data movement, final aggregation, and global ordering.

Jobs, stages, and tasks

Spark's UI uses several levels of execution vocabulary:

SQL/DataFrame execution:
  A high-level query or DataFrame action.

Job:
  Work launched by an action.

Stage:
  A portion of a job between shuffle boundaries.

Task:
  One attempt to process one partition within a stage.

A single high-level operation can create multiple jobs and stages. A stage with 2/2 tasks means two tasks completed; it does not mean that the dataset has two storage folders. A stage with 1/1 means one task completed. A skipped stage is not automatically a failure: Spark may not need to rerun an upstream stage for a particular job.

In the local application, two execution slots are available at a time because of local[2]. That limits concurrency, not the number of logical partitions Spark can plan.

What the tests prove

The learning repository's full regression suite passed after adding the Spark layout and join slices:

11 passed

The tests check the explicit schema, the twelve input rows, the daily revenue baseline, Parquet row preservation and date directories, the filtered partitioned read, the customer join, and the broadcast configuration boundary.

The test result proves that these invariants held for this fixture. It does not prove that the application is ready for a multi-node cluster, large shuffles, skew, schema evolution, or production recovery.

What this experiment taught

The important change from the DuckDB POC is not simply “Spark can run SQL-like code.” It is that execution becomes an inspectable distributed plan:

A logical transformation
  becomes a physical plan
  that creates jobs
  divided into stages
  executed as tasks
  over partitions.

The plan exposes where work remains local and where data moves. The Exchange operator is especially important: it marks a boundary where Spark must redistribute records before it can complete an operation such as a grouped aggregation, ordered output, or join.

What this local experiment does not prove

This was intentionally small and local. It does not demonstrate:

• cluster scheduling across machines;
• executor loss and task retry;
• shuffle spill to disk at scale;
• network saturation;
• skewed keys and straggler tasks;
• large-file or small-file production economics;
• schema compatibility across independently deployed producers;
• transactional output commits.

Those are later experiments, not assumptions to smuggle into this one.

Revisit checklist

Before moving on, I should be able to answer:

• What is the grain of the orders input and daily revenue output?
• Which Spark calls define a plan, and which calls trigger work?
• What is the difference between a job, stage, and task?
• Why does the aggregation introduce Exchange?
• Why does global ordering introduce another exchange and a sort?
• What does local[2] tell me, and what does it not tell me?
• Why is an explicit decimal schema safer for amount?
• Which values in this article were observed locally, and which are scale reasoning?

The next article moves from execution planning to physical storage: how Spark writes Parquet files, how directory partitioning enables pruning, and why more parallelism can create a small-files problem.