Data systems engineering POC 2: a beginner-friendly guide to Spark

Spark is often introduced with a short example:

df.groupBy("category").count().show()

That example is useful, but it leaves out the parts that matter when you operate a real data pipeline:

• Where does the data come from?
• What does Spark do before it returns a result?
• Why does one line of code create several jobs and stages?
• What is a partition?
• What is a shuffle?
• Why can two correct queries have very different costs?
• Where does the output live?
• How does a browser application display the result?
• What happens when the Spark process dies?

This project answers those questions by building a small application rather than by collecting disconnected code samples.

The application is called the GitHub Activity Observatory. It reads a bounded window of public GitHub activity, turns the raw events into typed analytical relations, writes Parquet data, and displays repository and commit activity in a local Streamlit application. GH Archive publishes public GitHub activity in hourly compressed JSON archives, with multiple event types and event-specific payloads.[1]

The dataset is not large enough to represent a production cluster by itself. The application is still shaped like a production data path, and its experiments make Spark's execution model visible on a laptop.

What we built

The end-to-end flow is:

GH Archive hourly JSON
        |
        v
Local compressed input
        |
        v
Spark batch process
        |
        +--------------------+
        |                    |
        v                    v
Bronze events          Silver push_commits
        |                    |
        +----------+---------+
                   v
             Gold reports
                   |
                   v
          Streamlit dashboard

The browser shows:

Most active repositories by events
Most active repositories by commits
Event-type distribution
Activity over time
Pipeline reconciliation
A sample of Silver commit rows
The selected Silver implementation
The formatted Silver physical plan

The user can also run the Spark batch process with its native Spark UI enabled. The browser dashboard answers “what did the data say?” The Spark UI answers “how did Spark produce it?”

The technology stack

Apache Spark is the processing engine. PySpark is its Python API, not a second processing engine.[2][3] Spark SQL and DataFrames are two structured ways to describe work that Spark can optimize and execute.[4]

The application repository

The learning project is:

data-platform-learning/02-spark-github-streamlit-app/

The first vertical slice contains these files:

02-spark-github-streamlit-app/
├── README.md
├── app/
│   └── streamlit_app.py
├── scripts/
│   ├── compare_silver_apis.py
│   ├── download_archive.py
│   ├── run_pipeline.py
│   └── run_app.sh
├── src/
│   └── github_activity/
│       ├── __init__.py
│       ├── bronze.py
│       ├── contracts.py
│       ├── gold.py
│       ├── pipeline.py
│       └── silver.py
├── tests/
│   ├── conftest.py
│   ├── fixtures/
│   │   ├── events.jsonl
│   │   └── silver_events.jsonl
│   ├── test_bronze_gold.py
│   └── test_silver.py
└── notes/
    └── silver-contract.md

Generated data is stored under data/, but that directory is ignored by Git:

data/raw/        downloaded GZIP archives
data/processed/  Bronze, Silver, Gold, and plan outputs
data/processed/metadata/  saved physical plans

The raw archive remains on the local machine. The source code, tests, contracts, and notes are the durable learning artifacts.

Running the application

The first verified archive is:

2015-01-01-15.json.gz

From the app directory:

cd data-platform-learning/02-spark-github-streamlit-app

Download an archive

uv run --no-sync python scripts/download_archive.py 2015-01-01 15

The downloader builds this URL:

https://data.gharchive.org/2015-01-01-15.json.gz

and stores the file under:

data/raw/2015-01-01-15.json.gz

The --no-sync flag is used in the verified checkout because Streamlit was installed into the local environment while the project lockfile was being reconciled. A fresh clone should eventually run with a fully synchronized uv.lock.

Run the Spark pipeline directly

uv run --no-sync python scripts/run_pipeline.py \
  --input data/raw/2015-01-01-15.json.gz \
  --output-root data/processed \
  --run-id first-real-archive \
  --silver-api dataframe

The real run produced:

records_read=11351
records_written=11351
records_parsed=11351
records_parse_failures=0
silver_normalization_api=dataframe
silver_push_commit_rows=10109
push_event_count=5815
push_events_with_commit_rows=5772
push_events_without_commit_rows=43

To use Spark SQL for the Silver normalization instead:

uv run --no-sync python scripts/run_pipeline.py \
  --input data/raw/2015-01-01-15.json.gz \
  --output-root data/processed \
  --run-id sql-real-archive \
  --silver-api sql

Start the browser application

bash scripts/run_app.sh

The app normally opens at:

http://localhost:8501

The sidebar lets you choose:

Archive input
DataFrame API or Spark SQL
Process data

The Streamlit app delegates Spark work to scripts/run_pipeline.py. It then reads small Gold Parquet reports with DuckDB and sends those small result tables to the browser. Streamlit reruns the Python script when a user interacts with widgets, which is why the Spark JVM is deliberately kept in a separate short-lived process rather than cached inside the browser server.[5]

Run Spark with the native UI

To inspect one Spark run directly:

uv run --no-sync python scripts/run_pipeline.py \
  --input data/raw/2015-01-01-15.json.gz \
  --output-root data/processed/execution-ui \
  --run-id execution-ui \
  --silver-api dataframe \
  --enable-ui \
  --keep-alive

The command prints the active Spark UI URL. Use the printed URL rather than assuming that port 4040 is free:

http://localhost:4040

The --keep-alive option keeps the Spark process alive while you inspect Jobs, Stages, SQL, and Executors. Press Enter in the terminal when finished.

What Spark is

Spark is not a programming language. It is a data-processing engine and a set of APIs for describing data-processing work.

You can use Spark through:

Scala
Java
Python through PySpark
R through SparkR
Spark SQL

The same Spark engine can be controlled by different front ends. In this project:

Our code:       Python
Our Spark API:  PySpark
Spark engine:   JVM-based Apache Spark
JVM runtime:    Java 17

PySpark does not replace Spark. It lets Python create Spark sessions, DataFrames, queries, and transformations. The Python driver communicates with the Spark JVM through the PySpark gateway. Spark then plans and executes the work.

A simplified view is:

Python application
        |
        v
PySpark API
        |
        v
Spark JVM driver
        |
        +--> optimizer
        +--> scheduler
        +--> local or cluster executors

In local mode, the application uses:

.master("local[2]")

This means one local application with two local execution threads. It gives us Spark's execution model without giving us a multi-node production cluster.

SparkSession, DataFrames, and plans

SparkSession is the main entry point for structured Spark work.

A DataFrame is not a pandas DataFrame. It is a distributed, schema-aware representation of rows and a computation plan.

For example:

completed = events.where(
    F.col("event_type") == "PushEvent"
)

This does not necessarily read all rows immediately. It adds a filter to the logical plan.

An action such as:

completed.count()

asks Spark to execute enough of the plan to produce a count.

The distinction is:

Transformation:
  Describes work.

Action:
  Requires a result and starts execution.

Common transformations include:

select
where/filter
withColumn
cast
join
groupBy/agg
explode
repartition
orderBy

Common actions include:

count
show
collect
write

Spark's laziness lets it see a larger section of the computation before choosing a physical strategy. It can push filters, remove unused columns, select a join strategy, insert exchanges, and combine operations.

The execution hierarchy

Spark uses several execution terms. They are related, but they are not synonyms.

Application
  A running Spark program.

SQL/DataFrame execution
  A high-level query or DataFrame action recorded by Spark.

Job
  Work launched by an action.

Stage
  A section of a job separated from another section by a shuffle boundary.

Task
  One attempt to process one partition within a stage.

Partition
  A slice of data or work processed by a task.

The relationship is not one-to-one:

One DataFrame action
  can create multiple Spark jobs.

One Spark job
  can contain multiple stages.

One stage
  can contain many tasks.

What happened in the real run

The labeled execution run reported:

Completed jobs: 31
Completed stages: 61
SQL executions: 11

The 11 high-level SQL executions corresponded to actions in the pipeline:

Each count and write is an action. Several actions can require multiple jobs because Spark may schedule multiple execution pieces around query stages, shuffles, and file writes.

There was also a separate Python RDD job for:

raw_lines.rdd.zipWithIndex()

That operation adds source_line_number to the raw input. It crosses into the RDD API and caused the job shown as:

runJob at PythonRDD.scala:218

That explains the difference between the 11 SQL executions and the 31 total Spark jobs.

Friendly action labels

Spark's default Python call sites are not very readable. Without labels, the Jobs tab contained names like:

$anonfun$withThreadLocalCaptured$1 at CompletableFuture.java:1814

The pipeline now wraps actions with:

with labeled_action(spark, "write_silver_push_commits"):
    push_commits.write.mode("overwrite").parquet(...)

The helper sets Spark job metadata through:

spark_context.setJobGroup(label, label)
spark_context.setJobDescription(label)
spark_context.setLocalProperty("callSite.short", label)

The UI can then show names such as:

count_bronze_events
write_bronze_events
count_silver_push_commits
write_silver_push_commits
write_gold_repository_commit_activity

The labels do not change Spark's work. They make the control-plane intent visible while the Jobs and Stages tabs continue to show the actual execution details.

Reading a physical plan

The daily report plan looked like this:

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

Read it from the bottom upward:

Scan csv:
  Read the input file.

Filter:
  Keep completed events/orders.

Project:
  Keep only columns needed downstream.

Partial HashAggregate:
  Aggregate locally before moving data.

Exchange:
  Redistribute data between partitions.

Final HashAggregate:
  Combine partial results.

Sort:
  Produce ordered output.

Exchange is especially important. It marks a point where Spark cannot complete the operation using only the current partition and must redistribute data.

Bronze: preserve the input envelope

The Bronze reader starts with:

spark.read.text(str(input_path))

Spark reads the local .json.gz file and decompresses it as it reads. Each newline-delimited JSON object becomes one DataFrame row.

The first archive produced:

11,351 Bronze event rows

The Bronze reader preserves:

source_file
source_line_number
raw_json
ingestion_run_id
event_id
event_type
event_time
is_public
actor_id
actor_login
repository_id
repository_name
payload_json
parse_status

The common fields are parsed with:

F.from_json(raw_json, COMMON_EVENT_SCHEMA)

The event-specific payload is preserved as JSON:

F.get_json_object("raw_json", "$.payload")

This is useful because different event types have different payload structures. A PushEvent payload is not the same shape as a PullRequestEvent payload.

Bronze answers:

What did the source send?
Which source file and line produced this row?
Can the common event envelope be parsed?

Silver: change the grain deliberately

The first Silver transformation focuses on PushEvent.

PushEvent is the top-level event type. It is not the same thing as payload.push_id.

A Bronze PushEvent can contain:

{
  "id": "event-123",
  "type": "PushEvent",
  "payload": {
    "commits": [{ "sha": "sha-a" }, { "sha": "sha-b" }]
  }
}

Bronze contains one row:

one row per GitHub event

Silver push_commits contains two rows:

one row per commit occurrence inside a PushEvent

The transformation is:

Filter PushEvent rows
  → parse payload_json
  → read commits[]
  → posexplode commits[]
  → carry parent context
  → write one row per commit occurrence

The Silver row includes:

event_id
repository_id
repository_name
actor_id
actor_login
event_time
commit_position
commit_sha
commit_message
commit_author_name
commit_author_email
commit_url
source_file
source_line_number
ingestion_run_id

The child key is:

(event_id, commit_position)

It is not only event_id, because one event can contain multiple commits.

Real Silver counts

The real archive contained:

Bronze events:                 11,351
PushEvents:                     5,815
PushEvents with commit rows:   5,772
PushEvents without commits:       43
Silver commit rows:            10,109

The 43 zero-commit PushEvents remain represented at the event level but create no rows in push_commits. A child table with no child objects should contain zero child rows; that is different from deleting the parent event.

Repeated commit SHAs

The archive contained:

10,109 commit rows
9,408 distinct commit SHAs
701 repeated SHA occurrences

The Silver table records commit occurrences inside push events. It does not silently deduplicate repeated SHAs. A globally deduplicated commit dimension would be a different data contract.

DataFrame API and Spark SQL

The DataFrame implementation uses Python method calls:

parsed_events = events.where(
    (F.col("event_type") == "PushEvent")
    & F.col("payload_json").isNotNull()
).withColumn(
    "push_payload",
    F.from_json(F.col("payload_json"), PUSH_PAYLOAD_SCHEMA),
)

The SQL implementation registers the same input as a temporary view and uses:

FROM bronze_events
LATERAL VIEW posexplode(push_payload.commits) exploded
    AS zero_based_position, commit_struct

The two implementations are different authoring styles:

DataFrame API:
  Python method calls describe the plan.

Spark SQL:
  SQL text describes the plan.

They are not different execution engines.

The real archive parity run produced:

dataframe_rows=10109
sql_rows=10109
schemas_equal=True
dataframe_only_rows=0
sql_only_rows=0

Both physical plans had the same meaningful shape:

Scan ExistingRDD
  → Filter
  → Project
  → Generate (posexplode)
  → Project

The attribute IDs and field aliases differed, but the business result and execution operators matched.

The plan began with Scan ExistingRDD because the Bronze reader temporarily crosses into the RDD API to add source line numbers. The SQL and DataFrame Silver functions received the same already-created Bronze DataFrame; they did not read the GZIP source independently during the parity comparison.

Parquet: file format versus partitioning

Parquet is the physical column-oriented file format used for the outputs. A Parquet file stores typed columns, compression metadata, row groups, and other information that readers can use to reduce work.

There are three different uses of “partition”:

Spark execution partition:
  A temporary unit of work processed by a task.

Storage partition:
  A directory such as event_date=2015-01-01.

Parquet file:
  A physical file stored inside a directory.

These are not the same thing.

The ecommerce physical-layout experiment demonstrated this with:

The six date directories came from six distinct order_date values. The unpartitioned file count followed the requested Spark output partition count.

With only twelve rows, twelve output files are a poor layout. With much larger data, a reasonable number of sufficiently sized files can improve parallel scan behavior. Too many tiny files create metadata, file-open, and planning overhead.

Predicate pushdown and partition pruning

A filtered Parquet read such as:

spark.read.parquet(path).where(
    "order_date = DATE '2026-03-01'"
)

can produce different plan evidence depending on the layout.

For unpartitioned data, the plan showed:

PushedFilters: order_date = 2026-03-01

Spark pushes the condition toward the file reader.

For directory-partitioned data, the plan showed:

PartitionFilters: order_date = 2026-03-01

Spark can use the directory name to avoid reading unrelated date partitions as input data.

In plain language:

Pushed filter:
  “Read the files, but filter as early as the reader allows.”

Partition filter:
  “Use the directory structure to skip unrelated groups first.”

Partitioning is a workload decision. A date column is often useful when queries filter by date. A high-cardinality field such as event_id or order_id can create too many small directories.

Joins: same logical result, different physical work

The ecommerce join experiment used:

orders:     12 rows
customers:   6 rows

The clean customer key was unique, so an inner join preserved one row per order:

12 orders × 1 matching customer = 12 joined rows

Spark's normal plan used:

BroadcastHashJoin Inner BuildRight
├── orders scan
└── BroadcastExchange
    └── customers scan

Because customers was small, Spark built a lookup structure from it and made that lookup available to the processing side. It avoided shuffling the orders input.

When automatic broadcast was disabled, the plan became:

SortMergeJoin Inner
├── Sort
│   └── Exchange
│       └── orders scan
└── Sort
    └── Exchange
        └── customers scan

Both sides were hash-partitioned by customer_id:

hash(customer_id) → one of 200 shuffle buckets

All rows with the same key use the same routing rule. Spark then sorts both sides and performs a sort-merge join.

Both strategies returned:

12 joined rows
12 distinct order IDs

The logical contract stayed the same. The physical work changed.

A duplicate customer key illustrates why cardinality must be checked before joining. If three orders match two customer rows, the join produces six matching pairs. A normal join does not merge or deduplicate those rows.

Skew: when one key dominates a partition

Hash partitioning keeps the same key together. That is necessary for a normal partitioned join, but it creates a problem when one key is unusually large.

The bounded skew probe added 1,000 deterministic orders for C001 to the original 12-order fixture:

total orders:       1,012
C001 rows:          1,003
shuffle buckets:      200

Observed distribution:

hot_customer_partition_ids=[180]
non_empty_partition_count=6
empty_partition_count=194
largest partition=180, rows=1003

Increasing the bucket count to 400 produced:

hot_customer_partition_ids=[180]
non_empty_partition_count=6
empty_partition_count=394
largest partition=180, rows=1003

The important lesson is:

More buckets do not split one key.

All C001 rows still have to go to the same hash bucket. Increasing the number of buckets can create more empty work without fixing the hot key.

Possible production mitigations include salting for suitable aggregations, pre-aggregation, separating high-volume keys, or using a different join strategy. Those are follow-up experiments; this POC verified the skew mechanism itself.

Failure boundary: the Spark JVM connection

The first Streamlit architecture cached a SparkSession inside the browser process:

@st.cache_resource
def get_spark():
    return SparkSession.builder...getOrCreate()

After the Spark JVM driver disappeared, Streamlit still held the Python object. The next rerun failed at:

spark.read.parquet(...)

with:

ConnectionRefusedError: [Errno 61] Connection refused

This was not a refusal from the Parquet filesystem. The stack trace reached:

spark._jsparkSession.read()

That is the Python-to-JVM Py4J boundary. The Spark JVM's local gateway port was dead.

The Spark log also reported:

RpcEndpointNotFoundException:
Cannot find endpoint: spark://CoarseGrainedScheduler@localhost:...

The application was changed so that:

Streamlit:
  Long-lived browser/control process.

Spark:
  Short-lived batch subprocess.

DuckDB:
  Reads small Gold Parquet reports for presentation.

This has two benefits:

1. A dead Spark JVM cannot remain hidden inside Streamlit's cached state.
2. The application boundary resembles a control plane launching a data-plane job.

The fix was verified by running the Streamlit control path after the Spark subprocess completed:

summary_records_read=11351
summary_records_written=11351
summary_parse_failures=0
repository_rows=6181

What the browser displays and what the Spark UI displays

These two interfaces answer different questions.

Streamlit dashboard

Streamlit shows the data product:

Bronze event count
PushEvent count
Silver commit count
PushEvents without commits
Repository event activity
Repository commit activity
Event-type distribution
Hourly activity
Silver sample rows

The chart calls are small and direct:

st.bar_chart(event_types.set_index("event_type"))
st.line_chart(hourly.set_index("event_hour"))

Streamlit provides chart elements for these tables.[6]

Spark UI

The Spark UI shows execution evidence:

Jobs:
  Which actions ran?

Stages:
  Where were shuffle boundaries?

Tasks:
  How many execution partitions were processed?

SQL/DataFrame:
  What physical operators did Spark choose?

Executors:
  What resources and task metrics were observed?

The dashboard does not replace the Spark UI. The Spark UI does not replace a row-level data viewer.

What was learned and what remains outside this POC

This foundation POC established:

SparkSession and local execution
PySpark versus Spark
DataFrame versus Spark SQL authoring
Lazy plans and actions
Jobs, stages, tasks, and exchanges
Explicit schemas
JSON parsing
Nested-array normalization
Silver grain and lineage
Parquet output
Storage partitioning
Predicate pushdown and pruning
Broadcast joins
Shuffle joins
Hash partitioning
Skew mechanism
Streamlit/Spark process boundaries

The following were deliberately not treated as prerequisites for closing this Spark foundation:

Full skew mitigation comparison
Malformed-data recovery matrix
Replay and idempotent output proof
Versioned run-evidence envelope
Architecture decision record
Cluster-scale capacity and cost model
Full multi-event Silver normalization

Those topics remain important, but they fit naturally into later event-log, lakehouse, trust-plane, and production-integration work. Adding another event parser is not the same as learning another Spark engine concept.

Revisit checklist

A reader should be able to explain these without opening the code:

• What is Spark, and what is PySpark?
• Why is Java required when the application is written in Python?
• What does SparkSession represent?
• Why is a DataFrame not the same as a pandas DataFrame?
• Which operations are transformations, and which are actions?
• Why can one action create multiple jobs?
• What separates one Spark stage from another?
• What does one task process?
• What does Exchange mean in a physical plan?
• Why did the pipeline create 31 jobs?
• Why did zipWithIndex() create a Python RDD job?
• What does a Parquet file represent?
• What is the difference between an execution partition and a storage partition?
• Why did six date directories remain six when Spark output partitions changed?
• What is the difference between a pushed filter and a partition filter?
• Why did Spark broadcast the customer table?
• Why does a shuffle join exchange both inputs?
• What does hashpartitioning(customer_id, 200) mean?
• Why does increasing the number of buckets not split one hot key?
• Why does one PushEvent become multiple Silver commit rows?
• Why do parent event fields repeat on those child rows?
• Why must event counts and commit counts use different grains?
• Why did the Streamlit app move Spark into a subprocess?
• Which facts came from actual execution, and which are scale reasoning?

Closing perspective

The most useful result of this POC is not a particular repository ranking. It is a working mental model of the path from application code to physical execution:

Python code
  → PySpark API
  → Spark logical plan
  → optimized physical plan
  → jobs
  → stages
  → tasks over partitions
  → Parquet outputs
  → Gold data product
  → browser view

The same business operation can be expressed in SQL or DataFrame code. The important questions remain the same:

What is the grain?
What is the key?
Where does data move?
What state is durable?
What proves correctness?
What happens when the job runs twice?
What happens when one key is much larger than the rest?

POC 2 answered those questions for bounded batch processing. POC 3 changes the problem: records will arrive over time, and the system will need event keys, partitions, offsets, retention, replay, event time, watermarks, state, CDC, and schema evolution.

Sources

[1] GH Archive official site — hourly public GitHub event archives and event payload format.

[2] Apache Spark overview — Spark engine, high-level APIs, local execution, and cluster concepts.

[3] PySpark API overview — PySpark as the Python API for Apache Spark.

[4] Spark SQL and DataFrames — structured processing APIs and Spark SQL optimization.

[5] Streamlit architecture — Streamlit execution flow and reruns.

[6] Streamlit chart APIs — built-in chart elements such as st.bar_chart and st.line_chart.