Build Dual-Engine JSON-to-Parquet

Transform nested JSON into matching Parquet tables with Polars and PySpark.

Introduction

30 Second Summary

A single online order can contain one customer plus several products. Turning that purchase history into clean totals becomes tricky when every detail arrives bundled together.

In this project, you will create the same local order-ingestion pipeline with Polars and PySpark. Both versions turn nested NDJSON events into matching orders plus item-level Parquet tables.

What You'll Build

Picture running either script to turn four nested order events into two analysis-ready tables whose counts and revenue totals match.

By the end of this project, you'll have:

  • A lazy Polars pipeline you can run to produce separate order and item-level Parquet files from nested purchase events.
  • A local PySpark pipeline that displays its explicit nested schema before writing equivalent Parquet datasets.
  • Parity checks that prove both engines produce 4 orders, 6 item rows, and 13,123 revenue cents.
  • Secret Mission: Add a business-quality branch that routes an order with a missing ID into a quarantine Parquet dataset.

Are there any prerequisites?

You'll need basic Python plus pandas-style DataFrame familiarity.

The setup expects a Mac with Visual Studio Code plus Java 17 or later. Step 1 checks for Python 3.10 or later before creating the isolated environment.

Before We Start

Before the hands-on work begins, commit to the shared data contract you are about to build. This keeps the reusable pipeline pattern clear while you implement it in two engines.

Set Up the Local Data Engineering Workspace

Your shared data contract only stays trustworthy when both engines read the same source files. A reproducible workspace gives each implementation the same starting point.

A project-specific Python environment isolates Polars 1.44.2 from other projects. It also pins PySpark 4.2.0 for the second implementation.

A local Spark check confirms Java 17+ can start the engine on your Mac.

In this step, get ready to:
  • Create the project folders plus the three starter files.
  • Confirm Python 3.10+ is available.
  • Prepare a virtual environment that passes the Polars and PySpark setup check.
Create the project structure

Each NDJSON line stores one complete JSON order event. The fixture gives both engines the same four nested orders.

  • Press Cmd+Space on macOS or the Windows key on Windows to open your search bar.
  • Type Visual Studio Code and press Enter.
  • Click File in the top menu bar.
  • Select Open Folder....
  • Select your Desktop in the folder picker.
  • Select New Folder.
  • Enter dual-engine-json-pipeline as the folder name.
  • Select Open on macOS or Select Folder on Windows.

Visual Studio Code may show a Workspace Trust dialog because this is a newly opened folder.

  • Select Yes, I trust the authors if the Workspace Trust dialog appears.

You should see dual-engine-json-pipeline at the top of the Explorer sidebar.

  • Right-click dual-engine-json-pipeline in the Explorer sidebar.
  • Select New Folder.
  • Enter data and press Enter.
  • Right-click dual-engine-json-pipeline again.
  • Select New Folder.
  • Enter scripts and press Enter.

You should now see data plus scripts under the project folder.

  • Right-click dual-engine-json-pipeline once more.
  • Select New Folder.
  • Enter output and press Enter.

The Explorer should now show the three folders that separate source data from scripts plus generated results.

  • Right-click dual-engine-json-pipeline in the Explorer sidebar.
  • Select New File.
  • Enter requirements.txt and press Enter.
  • Pin the project dependencies by copying this content into requirements.txt:
polars==1.44.2
pyspark==4.2.0

What do these pins control?

  • The polars==1.44.2 requirement gives the single-machine pipeline a fixed library release.
  • The pyspark==4.2.0 requirement gives the Spark implementation its matching pinned release.
  • Save requirements.txt with Cmd+S on macOS or Ctrl+S on Windows.
  • Check the Explorer sidebar for requirements.txt.

You should see requirements.txt at the top level of dual-engine-json-pipeline.

  • Right-click the data folder in the Explorer sidebar.
  • Select New File.
  • Enter orders.ndjson and press Enter.
  • Add the four nested order events by copying this content into data/orders.ndjson:
{"order_id":"o1001","event_time":"2026-09-01T10:00:00Z","customer":{"id":"c101","country":"NZ"},"items":[{"sku":"keyboard","quantity":2,"unit_price_cents":1999},{"sku":"cable","quantity":1,"unit_price_cents":950}]}
{"order_id":"o1002","event_time":"2026-09-01T10:05:00Z","customer":{"id":"c102","country":"AU"},"items":[{"sku":"mouse","quantity":1,"unit_price_cents":3900}]}
{"order_id":"o1003","event_time":"2026-09-01T10:10:00Z","customer":{"id":"c103","country":"NZ"},"items":[{"sku":"notebook","quantity":3,"unit_price_cents":525},{"sku":"pen-set","quantity":1,"unit_price_cents":1200}]}
{"order_id":"o1004","event_time":"2026-09-01T10:15:00Z","customer":{"id":"c104","country":"SG"},"items":[{"sku":"adapter","quantity":2,"unit_price_cents":750}]}

What does this fixture contain?

Each line represents one order event. Its customer value is a nested object.

Its items value is an array of purchased products. Both shapes remain nested until later pipeline steps normalize them.

  • Save data/orders.ndjson.
  • Check the Explorer sidebar for orders.ndjson inside data.

You should see orders.ndjson nested under the data folder.

  • Right-click the scripts folder in the Explorer sidebar.
  • Select New File.
  • Enter check_setup.py and press Enter.
  • Create the local engine check by copying this code into scripts/check_setup.py:
import polars as pl
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("dual-engine-setup-check")
    .getOrCreate()
)

spark.range(3).show()
print("Polars import succeeded:", pl.DataFrame({"ready": [True]}))
print("Local Polars and PySpark setup is ready.")
spark.stop()

What does this setup check do?

  • The imports prove that both pinned packages are available to the active Python environment.
  • The SparkSession configuration starts Spark in local mode.
  • The spark.range(3).show() call produces three rows with values from 0 through 2.
  • The spark.stop() call shuts down the local Spark session after the check completes.
  • Save scripts/check_setup.py.
  • Check the Explorer sidebar for check_setup.py inside scripts.

You should now see all three starter files in their expected locations.

Missing a file or folder?

  • Check that requirements.txt sits directly inside dual-engine-json-pipeline.
  • Check that orders.ndjson sits inside data.
  • Check that check_setup.py sits inside scripts.

Still stuck? Help me compare my Visual Studio Code project structure with the required files.

✔️ Awesome, I've got everything!

Your three starter files are saved in the project structure that both pipelines use.

ⓧ I'd like to double check the full code

Your requirements.txt should match:

polars==1.44.2
pyspark==4.2.0

Your data/orders.ndjson should match:

{"order_id":"o1001","event_time":"2026-09-01T10:00:00Z","customer":{"id":"c101","country":"NZ"},"items":[{"sku":"keyboard","quantity":2,"unit_price_cents":1999},{"sku":"cable","quantity":1,"unit_price_cents":950}]}
{"order_id":"o1002","event_time":"2026-09-01T10:05:00Z","customer":{"id":"c102","country":"AU"},"items":[{"sku":"mouse","quantity":1,"unit_price_cents":3900}]}
{"order_id":"o1003","event_time":"2026-09-01T10:10:00Z","customer":{"id":"c103","country":"NZ"},"items":[{"sku":"notebook","quantity":3,"unit_price_cents":525},{"sku":"pen-set","quantity":1,"unit_price_cents":1200}]}
{"order_id":"o1004","event_time":"2026-09-01T10:15:00Z","customer":{"id":"c104","country":"SG"},"items":[{"sku":"adapter","quantity":2,"unit_price_cents":750}]}

Your scripts/check_setup.py should match:

import polars as pl
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("dual-engine-setup-check")
    .getOrCreate()
)

spark.range(3).show()
print("Polars import succeeded:", pl.DataFrame({"ready": [True]}))
print("Local Polars and PySpark setup is ready.")
spark.stop()
Confirm the runtime requirement

Both pinned packages require Python 3.10 or later. Checking the interpreter first prevents confusing installation failures.

  • Select View in the Visual Studio Code menu bar.
  • Select Terminal to open the integrated terminal.
  • Display your installed Python version by running this command:
python3 --version

What does this command confirm?

The command prints the Python interpreter version available in your terminal. The first two version numbers must be 3.10 or higher for these package releases.

✔️ I see version 3.10 or higher

Your Python runtime meets the package requirement.

  • Keep the Visual Studio Code terminal open for the virtual environment commands.

ⓧ I see an older version

Your current interpreter is below the minimum supported version. Install the current stable macOS package before creating the environment.

  • Open the official Python downloads for macOS page.
  • Download the current stable macOS installer.
  • Complete the Python installation.
  • Close the current terminal session using the trash icon in the Terminal panel.
  • Select View in the Visual Studio Code menu bar.
  • Select Terminal to start a fresh terminal session.
  • Repeat the version check by running:
python3 --version

What confirms the upgrade?

The printed version should now be 3.10 or higher. That interpreter can create the project environment.

ⓧ Command not found

Your terminal cannot currently find the Python executable. Install the current stable macOS package to add it.

  • Open the official Python downloads for macOS page.
  • Download the current stable macOS installer.
  • Complete the Python installation.
  • Close the current terminal session using the trash icon in the Terminal panel.
  • Select View in the Visual Studio Code menu bar.
  • Select Terminal to start a fresh terminal session.
  • Check the new installation by running:
python3 --version

What confirms the installation?

The terminal should now print Python 3.10 or higher. This confirms the executable is available.

Build the isolated environment

A virtual environment keeps the pinned libraries inside this project. Activating it also makes later script commands use the same interpreter.

  • Create the project virtual environment in the terminal by running:
python3 -m venv .venv

What did this create?

The command creates an isolated environment in .venv inside dual-engine-json-pipeline. Its installed packages remain separate from your system Python.

  • Check the Explorer sidebar for the new .venv folder.

You should see .venv beside data plus scripts plus output.

  • Activate the new environment by running:
source .venv/bin/activate

What changed in the terminal?

Activation points Python package commands at the environment inside .venv. Your terminal prompt should now include the environment name.

You should see (.venv) near the start of the terminal prompt.

This installation can take a few minutes. The terminal may pause between progress updates while the packages are prepared.

  • Install the pinned dependencies from requirements.txt by running:
python -m pip install -r requirements.txt

What does pip install?

The requirements file installs Polars 1.44.2 plus PySpark 4.2.0 into the active environment. Future Python commands in this terminal use those pinned packages.

You should see the installation finish without an error. Keep (.venv) visible in the prompt.

Seeing an installation error?

  • Check that the terminal prompt includes (.venv) before repeating the installation.
  • Check that requirements.txt contains the two exact dependency pins from the full-code comparison.
  • Check that your Python version is 3.10 or higher.

Need another pair of eyes? Help me troubleshoot my pinned Polars and PySpark installation.

Before you run the setup check, which output signals do you expect from the two engines?

  • Start the local setup check by running:
python scripts/check_setup.py

What does this run prove?

  • The Spark table proves Java can start a local Spark session through PySpark.
  • The Polars DataFrame proves the pinned Polars package imports successfully.
  • The final message appears only after both checks complete.

You should see a three-row Spark table with values 0 through 2. You should then see Local Polars and PySpark setup is ready. after the Polars output.

That's your workspace ready. Both processing engines now start from the same isolated environment.

Spark does not start?

  • Check that Java 17 or later is installed.
  • Check that JAVA_HOME points to your installed Java distribution.
  • Check that the terminal prompt still includes (.venv).

Still blocked? Help me diagnose why the local Spark setup check cannot start.

Your local workspace can now start both engines from one isolated environment. Next, you'll inspect the nested order events with Polars.

Inspect Nested JSON with Polars

Your local workspace can now run both data engines. The next question is what shape the four order events have before any transformation begins.

Polars can scan NDJSON lazily. This first pass reveals whether the customer details and product items already behave like flat table columns.

In this step, get ready to:
  • Create a Polars script that lazily scans the order events.
  • Print the selected nested columns and their schema.
  • Identify why six products still appear inside four order rows.
Build a lazy view of the orders

A LazyFrame stores a query plan before loading the result into memory. Calling collect() executes that plan after the required columns have been selected.

  • Select the scripts folder in the Explorer sidebar.
  • Click the new-file icon in the Explorer toolbar.
  • Enter inspect_json.py as the file name.

You will see a blank inspect_json.py tab inside the existing scripts folder.

  • Add the input path and lazy inspection query by copying this code into scripts/inspect_json.py:
from pathlib import Path

import polars as pl

ROOT = Path(__file__).resolve().parents[1]
INPUT = ROOT / "data" / "orders.ndjson"

preview = (
    pl.scan_ndjson(INPUT)
    .select("order_id", "customer", "items")
    .collect()
)

print(preview)
print(preview.schema)

How Does the Lazy Inspection Work?

  • ROOT points to the dual-engine-json-pipeline folder from the script's location.
  • INPUT identifies the existing data/orders.ndjson source file.
  • pl.scan_ndjson(INPUT) creates the lazy scan without immediately materializing the table.
  • select("order_id", "customer", "items") limits the preview to the columns that reveal the nested structure.
  • collect() executes the query and returns the preview that the two print calls display.
  • Save scripts/inspect_json.py.
  • Confirm inspect_json.py remains listed under the scripts folder.

✔️ Awesome, I've got everything!

Good. Your saved inspection script now defines the source path and the lazy query that reveals the nested data shape.

ⓧ I'd like to double check the full code

Compare scripts/inspect_json.py with this complete reference:

from pathlib import Path

import polars as pl

ROOT = Path(__file__).resolve().parents[1]
INPUT = ROOT / "data" / "orders.ndjson"

preview = (
    pl.scan_ndjson(INPUT)
    .select("order_id", "customer", "items")
    .collect()
)

print(preview)
print(preview.schema)

This reference resolves the input path from the script's location. It delays execution until the query is collected.

Run the scan and inspect the shape

Running the script materializes the selected columns in your terminal. The preview gives you a direct view of how Polars represents each nested value.

Before you run the script, do you think the six products will appear as six rows or stay nested inside four order rows?

  • Display the lazy query result by running this command in the terminal from earlier:
python scripts/inspect_json.py

You should see four rows across the printed Polars table. The customer column contains nested struct values.

The items column contains list-shaped values made from item structs. The printed schema confirms that those lists remain inside the four order rows.

Why Aren't There Six Item Rows?

Each NDJSON line defines one order. Polars therefore preserves four source rows during this selection.

Each items value stores a list within its order row. Its item structs have not become independent table rows.

Don't See the Nested Preview?

  • Check that your terminal is inside the dual-engine-json-pipeline folder.
  • Confirm the activated environment from the previous step is still shown in your terminal prompt.
  • Confirm orders.ndjson remains inside the data folder.

Still stuck? Help me troubleshoot my nested Polars preview.

That scan exposed the modeling gap: four order rows still hold six products inside nested lists. Next, you will flatten those products into relational rows for Parquet output.

Normalize and Write Tables with Polars

Your inspection script proved that the four order events are readable. It also exposed the modeling gap because each items list still sits inside one order row.

This step resolves that gap with Polars. You will project the nested customer fields into an orders table. You will also explode the item structs into six relational rows before writing both tables as Parquet files.

In this step, get ready to:
  • Project order fields into an analysis-ready LazyFrame.
  • Explode nested item structs into relational rows with line totals.
  • Produce validated Parquet outputs for both tables.
Project the orders table

The source NDJSON stores customer details inside a struct. The orders projection gives those nested values their own columns while keeping the query lazy.

  • Select the scripts folder in the Visual Studio Code Explorer sidebar.
  • Click the New File button at the top of the Explorer sidebar.
  • Enter polars_pipeline.py as the file name.

You should see polars_pipeline.py listed inside the scripts folder. The empty editor tab is ready for the pipeline.

  • Build the path configuration and orders projection by copying this code into scripts/polars_pipeline.py:
from pathlib import Path

import polars as pl

ROOT = Path(__file__).resolve().parents[1]
INPUT = ROOT / "data" / "orders.ndjson"
OUTPUT = ROOT / "output" / "polars"


def main() -> None:
    OUTPUT.mkdir(parents=True, exist_ok=True)
    source = pl.scan_ndjson(INPUT)

    orders_lf = source.select(
        "order_id",
        "event_time",
        pl.col("customer").struct.field("id").alias("customer_id"),
        pl.col("customer").struct.field("country").alias("country"),
    )

What does this code do?

  • The three path constants locate the input data from the script. They also reserve output/polars for the generated tables.
  • The main() function creates the output folder before building the query.
  • pl.scan_ndjson(INPUT) represents the source as a lazy query. Polars can plan the transformations before reading the data.
  • orders_lf keeps the top-level order fields. The struct expressions extract customer_id and country as regular columns.
  • Save scripts/polars_pipeline.py.

The Explorer should continue to show polars_pipeline.py inside scripts. Your first lazy table definition is now stored in the project.

Can't find the new script?

  • Check that polars_pipeline.py appears inside the existing scripts folder.
  • Move the file into scripts if it was created beside the folder.

Still stuck? Help me place my Polars pipeline script in the correct project folder.

Normalize the item records

Each order currently carries a list of item structs. Exploding that list creates one row per product. Struct-field projections then turn each item property into a column.

  • Add the item transformation and output logic directly below the closing parenthesis of orders_lf:
    items_lf = (
        source.select("order_id", "items")
        .explode("items")
        .select(
            "order_id",
            pl.col("items").struct.field("sku").alias("sku"),
            pl.col("items").struct.field("quantity").alias("quantity"),
            pl.col("items").struct.field("unit_price_cents").alias("unit_price_cents"),
            (
                pl.col("items").struct.field("quantity")
                * pl.col("items").struct.field("unit_price_cents")
            ).alias("line_total_cents"),
        )
    )

    orders_df = orders_lf.collect()
    items_df = items_lf.collect()
    revenue_cents = items_df["line_total_cents"].sum()

    assert len(orders_df) == 4
    assert len(items_df) == 6
    assert revenue_cents == 13123

    orders_df.write_parquet(OUTPUT / "orders.parquet")
    items_df.write_parquet(OUTPUT / "order_items.parquet")


if __name__ == "__main__":
    main()

How does the normalization work?

  • explode("items") expands every item in an order into its own row. The four order events become six item rows.
  • struct.field() extracts each product property from the exploded struct.
  • line_total_cents multiplies quantity by unit price for each item row. Integer cents keep the validation exact.
  • collect() executes both lazy queries. The assertions enforce the expected row counts and revenue before either table is written.
  • write_parquet() stores the validated DataFrames as two columnar files under output/polars.
  • Save scripts/polars_pipeline.py.

Before you run the pipeline, how many order rows and item rows do you expect the assertions to accept?

  • Switch back to the activated terminal in Visual Studio Code.
  • Create the two Parquet outputs by running this command:
python scripts/polars_pipeline.py

You should see a new output/polars folder in the Explorer. It contains orders.parquet and order_items.parquet.

That is the modeling gap closed. Your nested events now exist as two analysis-ready tables.

Don't see the Parquet files?

  • Check the terminal for an assertion or syntax error that stopped the script before the write operations.
  • Confirm that scripts/polars_pipeline.py is inside the same dual-engine-json-pipeline folder as data/orders.ndjson.
  • Confirm that the activated terminal prompt still shows (.venv).

Need a hand? Help me debug why my Polars pipeline did not create its Parquet files.

Verify the data contract

The assertions protect the data contract inside the script. A final summary makes that contract visible whenever the pipeline succeeds.

  • Find the line if __name__ == "__main__": near the bottom of scripts/polars_pipeline.py.
  • Insert this output block directly above that line:
    print(orders_df)
    print(items_df)
    print(
        f"Polars checks passed: {len(orders_df)} orders, "
        f"{len(items_df)} items, {revenue_cents} revenue cents."
    )

What does this output show?

  • The first two calls print the collected orders and item DataFrames. You can inspect the normalized columns directly in the terminal.
  • The final message reports the values that already passed every assertion. It gives each successful run a compact acceptance result.
  • Save scripts/polars_pipeline.py.

Before you run the completed pipeline, what totals should appear if both tables still satisfy the contract?

  • Run the completed Polars pipeline from the activated terminal:
python scripts/polars_pipeline.py

You should see an orders table with 4 rows. You should also see an item table with 6 rows.

The final line should read Polars checks passed: 4 orders, 6 items, 13123 revenue cents. That confirmation proves both Parquet files were created from validated data.

Checks not passing?

  • Compare the items_lf expression with the full code below if the item count is incorrect.
  • Check that line_total_cents multiplies quantity by unit_price_cents if the revenue assertion fails.
  • Check that the print block remains indented inside main() if the terminal reports an indentation problem.

Still seeing a failure? Help me compare my Polars pipeline with the expected row counts and revenue.

✔️ Awesome, I've got everything!

Great work. Save scripts/polars_pipeline.py and keep both Parquet files under output/polars for the Spark implementation.

ⓧ I'd like to double check the full code

Compare your scripts/polars_pipeline.py with this complete version.

from pathlib import Path

import polars as pl

ROOT = Path(__file__).resolve().parents[1]
INPUT = ROOT / "data" / "orders.ndjson"
OUTPUT = ROOT / "output" / "polars"


def main() -> None:
    OUTPUT.mkdir(parents=True, exist_ok=True)
    source = pl.scan_ndjson(INPUT)

    orders_lf = source.select(
        "order_id",
        "event_time",
        pl.col("customer").struct.field("id").alias("customer_id"),
        pl.col("customer").struct.field("country").alias("country"),
    )

    items_lf = (
        source.select("order_id", "items")
        .explode("items")
        .select(
            "order_id",
            pl.col("items").struct.field("sku").alias("sku"),
            pl.col("items").struct.field("quantity").alias("quantity"),
            pl.col("items").struct.field("unit_price_cents").alias("unit_price_cents"),
            (
                pl.col("items").struct.field("quantity")
                * pl.col("items").struct.field("unit_price_cents")
            ).alias("line_total_cents"),
        )
    )

    orders_df = orders_lf.collect()
    items_df = items_lf.collect()
    revenue_cents = items_df["line_total_cents"].sum()

    assert len(orders_df) == 4
    assert len(items_df) == 6
    assert revenue_cents == 13123

    orders_df.write_parquet(OUTPUT / "orders.parquet")
    items_df.write_parquet(OUTPUT / "order_items.parquet")

    print(orders_df)
    print(items_df)
    print(
        f"Polars checks passed: {len(orders_df)} orders, "
        f"{len(items_df)} items, {revenue_cents} revenue cents."
    )


if __name__ == "__main__":
    main()

Your Polars pipeline now turns nested purchase events into two validated Parquet tables. Next, you will rebuild the same transformations with PySpark while preserving this exact contract.

Rebuild the Pipeline with PySpark

Your Polars pipeline already turns nested NDJSON into two Parquet outputs. Those files define the data contract that this second implementation must preserve.

This step expresses that contract again with PySpark DataFrames. An explicit nested schema lets you focus on the distributed transformation pattern using business logic you already understand.

In this step, get ready to:
  • Start local Spark with an explicit nested schema.
  • Normalize orders and item arrays with Spark DataFrames.
  • Display the results before writing two Spark Parquet datasets.
Define the nested schema

Spark can read the source with a DDL-formatted schema that describes every nested field. This gives the pipeline an explicit agreement about names and types before any transformations run.

  • In the Visual Studio Code Explorer sidebar, create pyspark_pipeline.py inside the existing scripts folder.
  • Define the imports, project paths, and nested schema by copying this code into scripts/pyspark_pipeline.py:
from pathlib import Path

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

ROOT = Path(__file__).resolve().parents[1]
INPUT = ROOT / "data" / "orders.ndjson"
POLARS_OUTPUT = ROOT / "output" / "polars"
SPARK_OUTPUT = ROOT / "output" / "spark"

SCHEMA = """
order_id STRING,
event_time STRING,
customer STRUCT<id: STRING, country: STRING>,
items ARRAY<STRUCT<sku: STRING, quantity: INT, unit_price_cents: INT>>
"""

What does this code do?

  • The imports provide filesystem paths, Spark column functions, and the local Spark session builder.
  • The four path constants keep the source data and both engines' outputs tied to the dual-engine-json-pipeline folder.
  • The SCHEMA string declares the top-level fields, the nested customer struct, and the array of item structs.
  • Save scripts/pyspark_pipeline.py.
  • Confirm pyspark_pipeline.py appears beneath the scripts folder in the Explorer sidebar.

Does the schema look incomplete?

  • Check that SCHEMA starts and ends with three double quotes.
  • Confirm the customer field uses a nested struct.
  • Confirm the items field uses an array containing item structs.

Still stuck? Help me check the nested schema in my PySpark pipeline.

The schema becomes useful when a Spark session applies it to the source file. The first runnable version prints the resulting tree so you can confirm that Spark understands the nesting.

  • Add the local Spark session and source reader below SCHEMA in scripts/pyspark_pipeline.py:
def main() -> None:
    spark = (
        SparkSession.builder
        .master("local[*]")
        .appName("dual-engine-json-pipeline")
        .getOrCreate()
    )

    source = spark.read.json(str(INPUT), schema=SCHEMA)

    source.printSchema()
    spark.stop()


if __name__ == "__main__":
    main()

How does the Spark reader work?

  • The session uses local[*] so Spark can run locally with the available processor cores.
  • The JSON reader applies SCHEMA while loading data/orders.ndjson.
  • The schema display exposes Spark's interpretation of the nested structs and array before the session stops.
  • Save scripts/pyspark_pipeline.py.

Before you run the script, what structure do you expect Spark to print for customer and items?

  • Return to the activated terminal from earlier and print the source schema by running:
python scripts/pyspark_pipeline.py

You should see a schema tree where customer is a struct. You should also see items as an array containing structs with product fields.

Spark session not starting?

  • Confirm the terminal still shows the activated .venv environment.
  • Check that Java is available to the same terminal session.
  • Confirm data/orders.ndjson remains inside the project folder.

Need a hand? Help me troubleshoot why my local PySpark session cannot read the NDJSON file.

Build the orders and items DataFrames

The Polars version already established the two table shapes. Spark now reproduces them with column selections plus an array explosion.

  • In scripts/pyspark_pipeline.py, find the line that creates source.
  • Create the orders DataFrame directly below that line by copying:
    orders = source.select(
        "order_id",
        "event_time",
        F.col("customer.id").alias("customer_id"),
        F.col("customer.country").alias("country"),
    )

How are nested customer fields projected?

  • The selection keeps the order identifier and event time as top-level columns.
  • The nested customer fields become the flat customer_id and country columns.
  • The aliases preserve the same orders-table contract produced by Polars.
  • Below source.printSchema(), add the orders display line:
    orders.show(truncate=False)

Why display the orders table?

The display provides a fast runtime check before any data is written. Full string values remain visible because truncation is disabled.

  • Save scripts/pyspark_pipeline.py.

Before you rerun the script, how many order rows do you expect after selecting one row per source event?

  • Display the flattened orders table by running:
python scripts/pyspark_pipeline.py

You should see four rows with order_id, event_time, customer_id, and country columns. The first half of the shared contract now works in Spark.

Orders table missing customer values?

  • Check that the nested paths are customer.id and customer.country.
  • Confirm the aliases match customer_id and country.
  • Make sure the orders selection sits inside main() after the source reader.

Still missing values? Help me debug my nested customer field selections in PySpark.

Each source event still holds its products inside one array. Exploding that array creates one row per item while preserving the parent order identifier.

  • Add the items transformation after the closing parenthesis of the orders selection:
    items = (
        source.select("order_id", F.explode("items").alias("item"))
        .select(
            "order_id",
            F.col("item.sku").alias("sku"),
            F.col("item.quantity").alias("quantity"),
            F.col("item.unit_price_cents").alias("unit_price_cents"),
            (
                F.col("item.quantity") * F.col("item.unit_price_cents")
            ).alias("line_total_cents"),
        )
    )

How does the item normalization work?

  • The explosion emits one row for each struct stored in the items array.
  • The second selection extracts the product fields from the temporary item struct.
  • The multiplication creates line_total_cents from quantity and unit price using integer cents.
  • Add the items display line directly below orders.show(truncate=False):
    items.show(truncate=False)

Why display the item rows?

The second display proves that Spark expanded the four source orders into individual product rows. It also exposes each calculated line total before the write.

  • Save scripts/pyspark_pipeline.py.

Before you rerun the script, do you expect the item table to contain four rows or six rows?

  • Display both normalized tables by running:
python scripts/pyspark_pipeline.py

You should see the four-row orders table followed by a six-row items table. Each item row should include its order identifier plus line_total_cents.

Items still appear nested?

  • Confirm the first selection applies F.explode("items") before selecting the struct fields.
  • Check that every product field starts with the temporary item alias.
  • Confirm the line total multiplies quantity by unit price.

Need help with the expansion? Help me debug why my PySpark items DataFrame does not show six flat rows.

Write the Spark Parquet datasets

The displayed DataFrames now match the shapes created by the Polars pipeline. Writing them to separate Spark dataset directories creates the second engine's persistent outputs.

  • Add both Parquet writes below items.show(truncate=False):
    orders.write.parquet(str(SPARK_OUTPUT / "orders"), mode="overwrite")
    items.write.parquet(str(SPARK_OUTPUT / "order_items"), mode="overwrite")

What do these writes create?

  • The first write stores the four order rows under output/spark/orders.
  • The second write stores the six item rows under output/spark/order_items.
  • Overwrite mode keeps repeated local runs predictable by replacing the previous datasets.
  • Save scripts/pyspark_pipeline.py.

Before you run the completed script, which two dataset directories do you expect Spark to create?

  • Run the completed Spark pipeline by using:
python scripts/pyspark_pipeline.py

Spark should print the explicit nested schema, a four-row orders table, and a six-row items table. The script should finish after writing both datasets under output/spark.

  • Refresh the Visual Studio Code Explorer sidebar.
  • Expand the existing output folder.
  • Confirm the new spark/orders and spark/order_items dataset directories are present.

Spark datasets not created?

  • Confirm both write lines appear before spark.stop() inside main().
  • Check that both destinations begin with SPARK_OUTPUT.
  • Look above the terminal prompt for the first message that identifies a failed Spark job or an invalid path.

Still no output? Help me troubleshoot why my PySpark Parquet datasets were not written.

That is the second engine working. Spark now reads the same events, produces the same table shapes, and persists them for comparison.

✔️ Awesome, I've got everything!

Great work. Make sure scripts/pyspark_pipeline.py is saved and both Spark dataset directories remain under output/spark.

ⓧ I'd like to double check the full code

Compare your scripts/pyspark_pipeline.py with this completed Step 4 version.

from pathlib import Path

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

ROOT = Path(__file__).resolve().parents[1]
INPUT = ROOT / "data" / "orders.ndjson"
POLARS_OUTPUT = ROOT / "output" / "polars"
SPARK_OUTPUT = ROOT / "output" / "spark"

SCHEMA = """
order_id STRING,
event_time STRING,
customer STRUCT<id: STRING, country: STRING>,
items ARRAY<STRUCT<sku: STRING, quantity: INT, unit_price_cents: INT>>
"""


def main() -> None:
    spark = (
        SparkSession.builder
        .master("local[*]")
        .appName("dual-engine-json-pipeline")
        .getOrCreate()
    )

    source = spark.read.json(str(INPUT), schema=SCHEMA)

    orders = source.select(
        "order_id",
        "event_time",
        F.col("customer.id").alias("customer_id"),
        F.col("customer.country").alias("country"),
    )

    items = (
        source.select("order_id", F.explode("items").alias("item"))
        .select(
            "order_id",
            F.col("item.sku").alias("sku"),
            F.col("item.quantity").alias("quantity"),
            F.col("item.unit_price_cents").alias("unit_price_cents"),
            (
                F.col("item.quantity") * F.col("item.unit_price_cents")
            ).alias("line_total_cents"),
        )
    )

    source.printSchema()
    orders.show(truncate=False)
    items.show(truncate=False)

    orders.write.parquet(str(SPARK_OUTPUT / "orders"), mode="overwrite")
    items.write.parquet(str(SPARK_OUTPUT / "order_items"), mode="overwrite")

    spark.stop()


if __name__ == "__main__":
    main()

Your Polars files and Spark datasets now represent the same intended contract. Next, you will read both outputs back and prove their row counts and revenue totals agree.

Prove Both Pipelines Agree

You have the PySpark transformation running now. It produces the same two table shapes as the Polars version from earlier.

Trust comes from a data contract that checks order counts, item counts, and revenue across independently produced tables. This step reloads all four Parquet outputs through Spark to prove that contract.

In this step, get ready to:
  • Add a reusable revenue helper for either engine's item output.
  • Reload both engines' Parquet outputs through Spark.
  • Validate the shared row-count and revenue contract.
Calculate revenue from either output

Each engine stores item revenue in the same column. A reusable helper can aggregate any Spark DataFrame containing line_total_cents.

  • Switch back to scripts/pyspark_pipeline.py from earlier.
  • Locate the closing """ immediately above def main() -> None:.
  • Insert the revenue helper between those two lines by adding this code:
def revenue_cents(dataframe) -> int:
    row = dataframe.agg(
        F.sum("line_total_cents").alias("revenue_cents")
    ).first()
    return row.revenue_cents

How Does the Helper Work?

  • The F.sum("line_total_cents") expression adds every line total in the supplied DataFrame.
  • The alias("revenue_cents") call gives the aggregated value a predictable field name.
  • The first() call retrieves the single aggregated row.
  • The row.revenue_cents expression returns the integer total that each parity assertion compares.
  • Save scripts/pyspark_pipeline.py.
  • Confirm the script still runs with the new helper by running this command in the activated terminal:
python scripts/pyspark_pipeline.py

What Does This Check Prove?

The command reruns the existing PySpark pipeline after Python loads the new helper. You should still see the nested schema followed by the four-row orders table and the six-row items table.

Does the Script Stop Early?

  • Check that revenue_cents() sits outside main() with no leading indentation.
  • Check that the helper appears after the closing quotes for SCHEMA.
  • Check that the terminal still shows the activated .venv environment.

Still stuck? Help me troubleshoot the revenue helper.

Read and compare every output

The parity block brings both engines' stored results into one Spark session. Assertions then turn the expected counts and revenue into executable acceptance checks.

  • In main(), find the existing items.write.parquet(str(SPARK_OUTPUT / "order_items"), mode="overwrite") line.
  • Insert the complete parity block directly below that line:
    polars_orders = spark.read.parquet(str(POLARS_OUTPUT / "orders.parquet"))
    polars_items = spark.read.parquet(str(POLARS_OUTPUT / "order_items.parquet"))
    spark_orders = spark.read.parquet(str(SPARK_OUTPUT / "orders"))
    spark_items = spark.read.parquet(str(SPARK_OUTPUT / "order_items"))

    assert polars_orders.count() == spark_orders.count() == 4
    assert polars_items.count() == spark_items.count() == 6
    assert revenue_cents(polars_items) == revenue_cents(spark_items) == 13123

    print("Parity checks passed: 4 orders, 6 items, 13123 revenue cents.")

What Do the Parity Checks Prove?

  • The four spark.read.parquet calls load both Polars files and both Spark datasets through the same engine.
  • The first assertion requires each orders output to contain exactly 4 rows.
  • The second assertion requires each item output to contain exactly 6 rows.
  • The final assertion requires both item outputs to total 13123 revenue cents.
  • The success message runs only after every assertion passes.
  • Keep the existing spark.stop() line directly below the new parity block.
  • Save scripts/pyspark_pipeline.py.
Run the final parity check

The completed script now writes fresh Spark outputs before reading every stored result back. The final message confirms that execution and data quality both succeeded.

Before you run the final check, do you expect the stored row counts and revenue totals to agree across both engines?

  • Verify the completed data contract by running this command in the activated terminal:
python scripts/pyspark_pipeline.py

What Should You See?

Spark first prints the explicit nested schema. It then displays the four-row orders table and the six-row items table.

The final line should read Parity checks passed: 4 orders, 6 items, 13123 revenue cents. This confirms that every parity assertion passed.

That is the trust check complete. Your two engines now prove the same four-order, six-item, 13,123-cent contract.

Do the Parity Checks Stop?

  • Confirm that output/polars/orders.parquet and output/polars/order_items.parquet still exist from the Polars pipeline.
  • Confirm that the Spark writes remain above the four Parquet readback lines.
  • Return to the Polars pipeline from Step 3 if an assertion reports different counts or revenue.
  • Check that JAVA_HOME still points to your Java installation if Spark cannot start.

Need another set of eyes? Help me diagnose the failed parity check.

✔️ Awesome, I've got everything!

Your saved script now reloads both engines' outputs and proves their counts and revenue agree.

ⓧ I'd like to double check the full code

  • Compare your saved scripts/pyspark_pipeline.py with this completed file:
from pathlib import Path

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

ROOT = Path(__file__).resolve().parents[1]
INPUT = ROOT / "data" / "orders.ndjson"
POLARS_OUTPUT = ROOT / "output" / "polars"
SPARK_OUTPUT = ROOT / "output" / "spark"

SCHEMA = """
order_id STRING,
event_time STRING,
customer STRUCT<id: STRING, country: STRING>,
items ARRAY<STRUCT<sku: STRING, quantity: INT, unit_price_cents: INT>>
"""


def revenue_cents(dataframe) -> int:
    row = dataframe.agg(
        F.sum("line_total_cents").alias("revenue_cents")
    ).first()
    return row.revenue_cents


def main() -> None:
    spark = (
        SparkSession.builder
        .master("local[*]")
        .appName("dual-engine-json-pipeline")
        .getOrCreate()
    )

    source = spark.read.json(str(INPUT), schema=SCHEMA)

    orders = source.select(
        "order_id",
        "event_time",
        F.col("customer.id").alias("customer_id"),
        F.col("customer.country").alias("country"),
    )

    items = (
        source.select("order_id", F.explode("items").alias("item"))
        .select(
            "order_id",
            F.col("item.sku").alias("sku"),
            F.col("item.quantity").alias("quantity"),
            F.col("item.unit_price_cents").alias("unit_price_cents"),
            (
                F.col("item.quantity") * F.col("item.unit_price_cents")
            ).alias("line_total_cents"),
        )
    )

    source.printSchema()
    orders.show(truncate=False)
    items.show(truncate=False)

    orders.write.parquet(str(SPARK_OUTPUT / "orders"), mode="overwrite")
    items.write.parquet(str(SPARK_OUTPUT / "order_items"), mode="overwrite")

    polars_orders = spark.read.parquet(str(POLARS_OUTPUT / "orders.parquet"))
    polars_items = spark.read.parquet(str(POLARS_OUTPUT / "order_items.parquet"))
    spark_orders = spark.read.parquet(str(SPARK_OUTPUT / "orders"))
    spark_items = spark.read.parquet(str(SPARK_OUTPUT / "order_items"))

    assert polars_orders.count() == spark_orders.count() == 4
    assert polars_items.count() == spark_items.count() == 6
    assert revenue_cents(polars_items) == revenue_cents(spark_items) == 13123

    print("Parity checks passed: 4 orders, 6 items, 13123 revenue cents.")
    spark.stop()


if __name__ == "__main__":
    main()

The helper calculates one revenue total for any item DataFrame. The final section reloads all four outputs before checking the shared contract and stopping Spark.

Secret mission

Quarantine an Invalid Order

Add an event with a null order ID to your source data. Build a PySpark quality branch that preserves the rejected record in a quarantine Parquet dataset while trusted records continue to their own output.

Clean Up Your Resources

Clean Up Your Resources

Everything in this project runs locally, so there are no ongoing cloud costs. Decide whether to keep the workspace available, pause it for another session, or delete it entirely.

Resources you used:

  • Project folder dual-engine-json-pipeline on your Desktop.
  • Project-specific Python environment at .venv with Polars 1.44.2 plus PySpark 4.2.0.
  • Source NDJSON files at data/orders.ndjson plus data/orders_with_invalid.ndjson.
  • Pipeline scripts inside the scripts folder.
  • Two Polars Parquet files at output/polars/orders.parquet plus output/polars/order_items.parquet.
  • PySpark output directories at output/spark/orders plus output/spark/order_items.
  • Secret Mission quality outputs at output/quality/valid_orders plus output/quality/quarantine.

Keep everything running

No action is needed. Choose this option if you want to rerun the pipelines or use their outputs in a later project.

  • Keep the dual-engine-json-pipeline folder on your Desktop.
  • Retain .venv so the pinned packages remain available.
  • Reuse the generated Parquet outputs for further analysis.

Each script stops its local Spark session after execution. Keeping the files does not leave a pipeline process running.

Pause - I'll come back to this later

Leave the project files untouched while you exit the isolated Python environment. This frees your current terminal for other work.

  • Deactivate the project environment by running:
deactivate

What does this command do?

This removes the virtual environment from your current shell. The installed packages remain inside .venv for your next session.

  • Check that the .venv prefix is no longer visible in your terminal prompt.

Your workspace is now paused. The source files, scripts, environments, and generated outputs remain ready for another session.

Can't deactivate the environment?

The command is available while the virtual environment is active. Return to the activated terminal from this project if your current shell does not recognize it.

Get help checking which Python environment is active.

Delete - I don't want to use this again

Remove all project resources to start fresh. Deleting the folder can feel final because this action cannot be undone.

  • Deactivate the project environment by running:
deactivate

What does this command do?

This detaches the virtual environment from your current shell. The environment remains inside the project folder until you delete that folder.

  • Move to your Desktop by running the first command below.
  • Delete the dual-engine-json-pipeline folder by running the second command below.
  • List your remaining Desktop contents by running the final command below.
cd ~/Desktop
rm -rf dual-engine-json-pipeline
ls

What do these commands do?

  • The first line moves your terminal to the Desktop.
  • The second line permanently removes the entire project folder.
  • The final line lists the remaining Desktop contents so you can confirm the result.

You should no longer see dual-engine-json-pipeline in the Desktop listing. That completes the cleanup of every local source file, script, environment, and output from this project.

Still see the project folder?

  • Check that the folder name is exactly dual-engine-json-pipeline.
  • Check that your terminal has permission to remove files from the Desktop.
  • Get help removing the local project safely.

Nice Work!

Nice Work!

What an achievement! Your Polars and PySpark pipelines turn nested NDJSON orders into analysis-ready Parquet outputs under one data contract.

You've learned how to:

  • Build an orders table with a lazy DataFrame that projects fields from nested customer structs.
  • Rebuild the normalization with an explicit nested schema. Apply array explosion to produce six rows in the order_items table.
  • Prove output equality with automated parity checks. Both engines report 4 orders. They also report 6 items. Their revenue total is 13,123 cents.
  • Complete the optional Secret Mission by adding a PySpark quality gate. The gate routes the null-order-ID event into a quarantine dataset while preserving the four valid orders.

Ready to quiz yourself?