Build a Book Price Pipeline

Scrape book listings into a local Kafka and Spark pipeline with a dashboard.

Introduction

30 Second Summary

A library can fill up while a shuttle stop gets busy within minutes. Static reports leave campus teams reacting to patterns that have already changed.

In this project, you will build Campus Pulse as a local data pipeline for campus operations events. A single workflow turns live activity into a compact summary for operational decisions.

What You'll Build

You'll trigger one Airflow workflow that launches Spark to turn fresh events from a Kafka topic into a queryable JSON summary for campus zones.

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

  • You can publish six campus events through the campus-events topic. A replay displays every JSON record.
  • A bounded Spark Structured Streaming job drains all current events before stopping. Its typed bronze data stores each field in Parquet format for queries.
  • A two-stage Airflow workflow runs ingestion before aggregation. Its JSON activity summary reports totals by campus zone and event type.
  • Secret Mission: Add a quality gate that diverts events with unknown campus zones into a quarantine dataset.

Are there any prerequisites?

Basic Python and SQL knowledge is expected. You should also feel comfortable using Terminal on macOS.

Before We Start

Campus Pulse supports a campus decision with a measurement built from fresh activity data. Before any hands-on work, commit to that purpose in one sentence.

Set Up the Local Streaming Stack

Your Campus Pulse pipeline now supports a clear campus decision. A reliable demo also needs compatible runtimes plus a reachable broker before events can produce visible results.

This step prepares a local workspace in VS Code with Python 3.12. You will connect Apache Spark 4.2.0. You will also prepare Apache Airflow 3.3.1 plus Apache Kafka 4.3.1 through Docker Compose.

In this step, get ready to:
  • Create the Campus Pulse workspace with an isolated Python environment.
  • Install the pinned Spark runtime plus Airflow packages.
  • Start Kafka with an empty topic and verify the complete stack.
Create the project workspace and Python environment

The project needs separate locations for source events, Spark jobs, Airflow workflows, and generated output. A virtual environment keeps the project's Python packages isolated from other work on your Mac.

  • Move to your Desktop and create the project directories by running these commands:
cd ~/Desktop
mkdir -p campus-pulse/{data,jobs,dags,output}
cd campus-pulse
ls

What Did These Commands Create?

  • The campus-pulse directory keeps every project artifact together.
  • The data directory will hold event batches.
  • The jobs directory will hold Spark jobs.
  • The dags directory will hold the Airflow workflow.
  • The output directory will hold checkpoints plus generated datasets.

You should see dags, data, jobs, and output in the terminal output.

Directories Missing?

Confirm your terminal prompt points to the campus-pulse directory. If the list is incomplete, rerun the directory creation command from your Desktop.

If the shell rejects the directory command, help me create the Campus Pulse directory structure.

  • Press Cmd+Space on macOS or the Windows key on Windows to open the system search bar.
  • Type VS Code and press Enter to launch it.
  • Use the folder picker in VS Code to select the campus-pulse folder on your Desktop.
  • Open an integrated terminal from the terminal menu in VS Code.

Your new terminal starts inside the campus-pulse directory. The project folders are visible in the VS Code file sidebar.

Python 3.12 gives Airflow a supported interpreter while keeping this project on a reproducible version. Check the available interpreter before creating the environment.

  • Check for Python 3.12 by running this command:
python3.12 --version

What Does This Check?

The command asks the python3.12 executable to report its version. The result determines whether you can create the project's environment immediately.

✔️ I see version 3.12

Python 3.12 is available. Continue below to create the project environment.

ⓧ I see another version

The executable is resolving to an unexpected Python release. Use Homebrew to install the required formula.

  • Install the Python 3.12 formula by running this command:
brew install python@3.12

What Does This Command Do?

Homebrew installs the versioned Python formula. The formula provides the python3.12 executable used below.

ⓧ Command not found

Python 3.12 is not available through your current shell. Homebrew can add the required executable.

  • Install Python 3.12 by running this command:
brew install python@3.12

What Does This Command Do?

Homebrew downloads the versioned Python formula. It exposes python3.12 for creating the isolated environment.

  • Create the project environment and activate it by running these commands:
python3.12 -m venv .venv
source .venv/bin/activate
python --version

How Does the Environment Work?

  • The first command creates .venv with Python 3.12.
  • The second command places that environment first in your terminal path.
  • The final command confirms which Python interpreter the terminal now uses.

You should see Python 3.12 in the final output. Your terminal prompt should also include (.venv).

Environment Not Activating?

Confirm that the terminal is inside campus-pulse. The activation script exists at .venv/bin/activate only after the environment creation command succeeds.

If the prompt still does not show the environment, help me activate my Python virtual environment.

Install Spark and Airflow

Spark processes the Kafka records in later steps. Airflow eventually launches those Spark jobs in a defined order.

Both tools are version-pinned because their command interfaces and package dependencies change across releases. Start by checking whether the required Spark runtime is already available.

  • Check the installed Spark version by running this command:
spark-submit --version

What Does This Check?

The spark-submit command reports the Spark runtime available in your terminal. Campus Pulse uses version 4.2.0 throughout its streaming jobs.

✔️ I see version 4.2.0

Spark 4.2.0 is ready. Continue below to verify it from the project environment.

ⓧ I see an older version

Your installed Spark formula needs an upgrade before it can match this project. Homebrew can update that formula without changing the project files.

  • Upgrade the Apache Spark formula by running this command:
brew upgrade apache-spark

What Does This Command Do?

Homebrew upgrades the installed Apache Spark formula to its current stable release. The formula provides the spark-submit executable.

ⓧ Command not found

Apache Spark is not installed through your current Homebrew setup. Install the formula before continuing.

  • Install Apache Spark by running this command:
brew install apache-spark

What Does This Command Do?

Homebrew installs Apache Spark plus its required dependencies. The installed formula includes spark-submit for running the jobs you build later.

  • Confirm the active Spark runtime by running the version check again:
spark-submit --version

What Should This Confirm?

The version banner confirms that the terminal can reach Spark. It should identify release 4.2.0.

Spark 4.2.0 is now available for local job submission.

Spark Version Still Wrong?

Restart the integrated terminal after the Homebrew command completes. This gives the shell a fresh view of the installed executable.

If an older release still appears, help me resolve my Spark command path.

Airflow belongs inside the active .venv environment. Check its current state before installing the pinned release.

  • Check whether Airflow is available in the active environment by running this command:
airflow version

What Does This Check?

The command reports the Airflow release available to the active environment. Campus Pulse requires version 3.3.1.

✔️ I see version 3.3.1

Airflow 3.3.1 is already available in the active environment. Continue below to pin the standard provider.

ⓧ I see an older version

The active environment contains a different Airflow release. Reinstall the required version with its Python 3.12 constraints file.

  • Install the pinned Airflow release by running this command:
pip install "apache-airflow==3.3.1" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.3.1/constraints-3.12.txt"

Why Use a Constraints File?

The constraints file selects a dependency set tested for Airflow 3.3.1 on Python 3.12. This reduces version conflicts inside the environment.

ⓧ Command not found

The new virtual environment does not contain Airflow yet. Install the required release with its matching Python constraints.

  • Install Airflow 3.3.1 by running this command:
pip install "apache-airflow==3.3.1" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.3.1/constraints-3.12.txt"

Why Use a Constraints File?

The constraints file gives pip a tested dependency set for Airflow 3.3.1 on Python 3.12. The result is a reproducible local installation.

The standard provider supplies the operator used by the workflow later. Its version is pinned to 1.17.0 for compatibility with this project.

This installation can take several minutes while pip resolves and installs the provider dependencies. A long stream of package output is normal.

  • Install the standard provider and check the environment by running these commands:
pip install "apache-airflow-providers-standard==1.17.0"
pip check

What Do These Commands Prove?

  • The first command pins the standard provider to version 1.17.0.
  • The pip check command checks the installed packages for dependency conflicts.

The final check should report that the environment has no broken package requirements. That confirms the Airflow installation is internally consistent.

Airflow Installation Failing?

Confirm that (.venv) appears in your terminal prompt. Also confirm that python --version reports Python 3.12.

If dependency resolution still fails, help me debug my pinned Airflow installation.

Start Kafka and verify the stack

A Kafka topic is the durable handoff point between event producers and Spark. Docker Compose gives the project one local broker at localhost:9092.

  • Use the VS Code file sidebar to create compose.yaml inside the campus-pulse folder.
  • Define the local Kafka service by pasting this configuration into compose.yaml:
services:
  kafka:
    image: apache/kafka:4.3.1
    ports:
      - "9092:9092"

What Does This Configuration Do?

  • The kafka service runs the official apache/kafka:4.3.1 image.
  • The port mapping exposes the broker on localhost:9092 for Spark running on your Mac.
  • Save compose.yaml.
  • Confirm that compose.yaml appears beside the project directories in the VS Code file sidebar.

Configuration File Missing?

Confirm that the file is named compose.yaml without an added text extension. It must sit directly inside campus-pulse.

If VS Code placed it inside another directory, help me fix my Compose file location.

✔️ Awesome, I've got everything!

Your saved Compose configuration defines the pinned Kafka service and its local port mapping.

ⓧ I'd like to double check the full code

  • Compare your complete compose.yaml file with this reference:
services:
  kafka:
    image: apache/kafka:4.3.1
    ports:
      - "9092:9092"

What Should Match?

The image version plus the port mapping should match this file exactly. Consistent spacing keeps the YAML structure valid.

Docker Desktop must be running before Compose can create the Kafka container. Its first startup can take a minute while the local Docker engine becomes ready.

  • Press Cmd+Space on macOS or the Windows key on Windows to open the system search bar.
  • Type Docker Desktop and press Enter to launch it.
  • Wait for Docker Desktop to report that its engine is ready.
  • Start the Kafka service from the VS Code terminal by running this command:
docker compose up -d

What Does This Command Do?

Docker Compose reads compose.yaml and starts the Kafka service in detached mode. Detached mode returns control of the terminal while the broker continues running.

You should see Compose create the local resources and start the kafka service. The first run can take longer because Docker downloads the pinned image.

Kafka Service Not Starting?

Confirm that Docker Desktop has finished starting. Also confirm that your terminal is inside the folder containing compose.yaml.

If the service still fails, help me debug my local Kafka Compose service.

The running broker now needs the empty campus-events topic that receives project events in the next step.

  • Create the campus-events topic inside the Kafka service by running this command:
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --create --topic campus-events --bootstrap-server :9092

What Does This Command Do?

Docker Compose runs Kafka's topic tool inside the existing kafka service. The tool connects to the broker and creates campus-events.

The terminal should confirm creation of the campus-events topic. No producer has published records yet, so the topic is empty.

Topic Creation Failing?

Give the broker a few seconds to finish starting before rerunning the topic command. Confirm that the Compose service is named kafka in compose.yaml.

If the broker remains unreachable, help me create the campus-events topic.

Before you run the final checks, predict which command proves that the broker is reachable.

  • Verify Spark, Airflow, and the Kafka cluster by running these commands:
spark-submit --version
airflow version
docker compose exec kafka /opt/kafka/bin/kafka-cluster.sh cluster-id --bootstrap-server :9092

What Do These Checks Prove?

  • The first command confirms that Spark is available to submit local processing jobs.
  • The second command confirms that the active Python environment contains the required Airflow release.
  • The third command connects to Kafka and requests the local cluster identifier.

You should see Spark report 4.2.0. Airflow should report 3.3.1. Kafka should print a cluster ID.

One of the Final Checks Failed?

Use the failing command to isolate the layer that needs attention. A Spark failure points to Homebrew, an Airflow failure points to the active .venv, and a Kafka failure points to Docker Compose.

For a guided diagnosis, help me debug my local streaming stack verification.

That is the foundation in place. Next, you will publish six campus events and watch Kafka return every record.

Publish and Inspect Campus Events

Your local Apache Kafka broker is live. Campus Pulse can now accept events from a producer.

A durable handoff keeps events available for downstream processors. This step proves that handoff with six JSON campus records.

In this step, get ready to:
  • Create a repeatable six-record campus event batch.
  • Publish the event batch to the campus-events topic.
  • Read every retained event from the start of the topic.
Create the campus event batch

An event schema gives every record the same named fields. This batch covers the three campus locations used by the later pipeline.

  • Create events_batch_1.jsonl inside the data folder using the file creation control in VS Code's Explorer sidebar.
  • Fill data/events_batch_1.jsonl by pasting these records:
{"event_id":"evt-001","event_time":"2026-09-03T08:00:00Z","zone":"library","event_type":"entry","count":1}
{"event_id":"evt-002","event_time":"2026-09-03T08:00:15Z","zone":"gym","event_type":"entry","count":1}
{"event_id":"evt-003","event_time":"2026-09-03T08:01:00Z","zone":"library","event_type":"checkout","count":2}
{"event_id":"evt-004","event_time":"2026-09-03T08:01:30Z","zone":"north_shuttle_stop","event_type":"arrival","count":14}
{"event_id":"evt-005","event_time":"2026-09-03T08:02:00Z","zone":"library","event_type":"entry","count":1}
{"event_id":"evt-006","event_time":"2026-09-03T08:02:15Z","zone":"gym","event_type":"entry","count":1}

How Is Each Event Structured?

  • The event_id field identifies one campus event.
  • The event_time field records when the activity happened.
  • The zone field names the campus location.
  • The event_type field describes the activity.
  • The numeric count field measures the activity volume.
  • Save data/events_batch_1.jsonl.
  • Confirm events_batch_1.jsonl contains six complete lines.

File Showing Fewer Than Six Lines?

  • Compare the file with the full-code tab below.
  • Check that each event occupies one complete line.
  • Ask for help to fix my JSONL event batch.

Your first pipeline payload is ready for launch.

✔️ Awesome, I've got everything!

The file now matches the six-record batch used throughout the pipeline.

ⓧ I'd like to double check the full code

{"event_id":"evt-001","event_time":"2026-09-03T08:00:00Z","zone":"library","event_type":"entry","count":1}
{"event_id":"evt-002","event_time":"2026-09-03T08:00:15Z","zone":"gym","event_type":"entry","count":1}
{"event_id":"evt-003","event_time":"2026-09-03T08:01:00Z","zone":"library","event_type":"checkout","count":2}
{"event_id":"evt-004","event_time":"2026-09-03T08:01:30Z","zone":"north_shuttle_stop","event_type":"arrival","count":14}
{"event_id":"evt-005","event_time":"2026-09-03T08:02:00Z","zone":"library","event_type":"entry","count":1}
{"event_id":"evt-006","event_time":"2026-09-03T08:02:15Z","zone":"gym","event_type":"entry","count":1}
Publish and read the events

Each line in the file becomes one Kafka record. The topic preserves those records for the consumer check.

  • Publish the six records to campus-events by running this command in the terminal from earlier:
docker compose exec -T kafka /opt/kafka/bin/kafka-console-producer.sh --topic campus-events --bootstrap-server localhost:9092 < data/events_batch_1.jsonl

How Does the Batch Reach Kafka?

Docker Compose runs the console producer inside the Kafka service. The no-TTY option lets the file flow through standard input.

Each line becomes one record on campus-events.

Producer Did Not Finish?

  • Confirm the kafka service is still running in Docker Desktop.
  • Check that your terminal is still inside campus-pulse.
  • Ask for help to debug the Kafka producer command.

Pause for one prediction: how many records the consumer will print.

  • Read the topic from its earliest retained record by running this command:
docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh --topic campus-events --from-beginning --bootstrap-server localhost:9092

What Does This Command Do?

The console consumer reads records stored in campus-events. The --from-beginning option starts with the topic's earliest retained record.

The consumer remains attached after printing the current records. You can stop it after verifying the complete batch.

You will see all six JSON lines from data/events_batch_1.jsonl. The final line contains evt-006 for a gym entry.

  • Stop the consumer after the sixth record by pressing Control-C.

Missing One of the Six Records?

  • Confirm the producer command completed before you started the consumer.
  • Check that data/events_batch_1.jsonl still contains all six records.
  • Ask for help to find my missing Kafka records.

That is the broker handoff proven: your six campus events survive after the producer finishes. Next, you will connect Spark Structured Streaming and inspect the raw payload Spark receives.

Inspect Raw Kafka Payloads

Your six campus events are now readable from Kafka. That proves the broker can carry each record into the next processing stage.

The next question is whether Apache Spark can use each payload for calculations.

In this step, you'll connect Structured Streaming to the topic. Then you'll inspect the shape Spark receives.

In this step, get ready to:
  • Create a local Spark session for campus event ingestion.
  • Subscribe to the campus events through a streaming source.
  • Drain the current records into a console table to inspect their shape.
Connect Spark to the campus topic

A SparkSession is the entry point for reading data with Spark. The local master lets the job use the logical processor cores on your Mac.

  • In the VS Code file tree, select the jobs folder inside campus-pulse.
  • Create stream_events.py inside the jobs folder.
  • Add the local Spark session by pasting this code into stream_events.py:
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("CampusEventIngest")
    .getOrCreate()
)

What Does This Code Do?

  • The SparkSession import provides Spark's main Python entry point.
  • The spark variable holds the session used by the rest of the job.
  • The local[*] master uses the logical processor cores available on your Mac.
  • The CampusEventIngest name identifies this application in Spark logs.
  • Save jobs/stream_events.py.
  • Confirm stream_events.py appears inside the jobs folder in the VS Code file tree.

File in the Wrong Folder?

  • Confirm the file path is jobs/stream_events.py inside campus-pulse.
  • Compare the capitalization of SparkSession with the code block.

If the editor still reports a problem, help me check my Spark session code.

Kafka supplies each record through a binary value field. This source definition tells Spark which broker contains the records.

  • Add the streaming source below the spark block by pasting this code:
raw_events = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "campus-events")
    .option("startingOffsets", "earliest")
    .load()
)

How Does the Source Work?

  • The readStream property creates a streaming data source.
  • The kafka format selects Spark's Kafka connector.
  • The kafka.bootstrap.servers option points to the local broker at localhost:9092.
  • The subscribe option selects the campus-events topic.
  • The startingOffsets option begins the first run at the earliest available records.
  • Save jobs/stream_events.py.

The first package resolution can take a little longer while Spark downloads the Kafka connector.

  • Check that Spark can build the source definition by running this command:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0 jobs/stream_events.py

What Does This Check Prove?

Spark loads the Kafka connector package. It also builds the source definition before returning you to the Terminal prompt.

Spark Did Not Return to the Prompt?

  • Confirm your internet connection can reach the package repositories used by Spark.
  • Confirm your activated .venv remains active in the current Terminal.
  • Compare the connector coordinate with org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0.

If the command still stalls or exits with an error, help me debug the Spark Kafka connector startup.

Drain the records into the console

An Available Now trigger processes every record currently waiting in the topic. The query terminates after that bounded batch finishes.

  • Add the console query immediately below the raw_events block by pasting this code:
query = (
    raw_events
    .selectExpr("CAST(value AS STRING) AS json_value")
    .writeStream
    .format("console")
    .trigger(availableNow=True)
    .start()
)

query.awaitTermination()
spark.stop()

What Does the Query Do?

  • The selectExpr expression casts each binary Kafka value to a readable string named json_value.
  • The console format prints the query result in the Terminal.
  • The availableNow=True trigger drains all currently available records.
  • The awaitTermination() call keeps the script active until processing finishes.
  • The spark.stop() call closes the local Spark session.
  • Save jobs/stream_events.py.

✔️ Awesome, I've got everything!

Your three code sections now form the complete starter stream.

ⓧ I'd like to double check the full code

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("CampusEventIngest")
    .getOrCreate()
)

raw_events = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "campus-events")
    .option("startingOffsets", "earliest")
    .load()
)

query = (
    raw_events
    .selectExpr("CAST(value AS STRING) AS json_value")
    .writeStream
    .format("console")
    .trigger(availableNow=True)
    .start()
)

query.awaitTermination()
spark.stop()

What Does This Reference Show?

This reference shows the complete starter file in execution order. The session creates the source before the bounded query prints its result.

Before you run this, do you expect Spark to expose count as its own column?

  • Run the bounded streaming query by entering this command:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0 jobs/stream_events.py

What Should You See?

Spark drains the six records before the script exits. You'll see a console table with one json_value column containing a complete JSON string in each row.

The count field remains embedded inside each string. Spark cannot sum that field as a typed numeric column in this shape.

No Campus Records in the Table?

  • Confirm the Kafka service from earlier remains running in Docker Desktop.
  • Compare the broker address with localhost:9092.
  • Compare the subscribed topic with campus-events.

If the query still cannot read the records, help me debug my bounded Spark Kafka query.

You found the exact boundary between transporting an event and understanding its fields. Next, you'll turn each raw payload into typed columns that Spark can aggregate.

Build the Typed Activity Data Product

Your first Apache Spark stream reached Apache Kafka in the previous step. Each record still arrived as one opaque JSON string. That blocked numeric aggregation with count.

Explicit parsing gives every event field a type. A checkpoint records consumed offsets. Parquet preserves the typed bronze data. You will use that bronze dataset to build a queryable activity summary.

In this step, get ready to:
  • Parse Kafka JSON into typed event columns.
  • Persist incremental bronze records with checkpointed Parquet output.
  • Aggregate activity counts into a JSON summary.
Parse and checkpoint the stream

Spark Structured Streaming reads Kafka value as binary data. The current cast exposes readable JSON text. from_json uses a schema to turn that text into named columns with usable types.

  • Switch back to jobs/stream_events.py in VS Code.
  • Replace the current from pyspark.sql import SparkSession line with this configuration block:
from pathlib import Path

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

PROJECT_ROOT = Path(__file__).resolve().parents[1]
BRONZE_PATH = str(PROJECT_ROOT / "output" / "bronze")
CHECKPOINT_PATH = str(PROJECT_ROOT / "output" / "checkpoints" / "bronze")
EVENT_SCHEMA = "event_id STRING, event_time STRING, zone STRING, event_type STRING, count INT"

What Does This Configuration Do?

  • PROJECT_ROOT anchors every generated path to the campus-pulse folder.
  • BRONZE_PATH identifies where Spark stores typed event records.
  • CHECKPOINT_PATH identifies where the stream stores its offset progress.
  • EVENT_SCHEMA assigns a name and data type to every field in the event envelope.
  • Save jobs/stream_events.py.
  • Return to the Terminal from earlier.
  • Confirm the existing stream still reaches Kafka by running this command:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0 jobs/stream_events.py

What Does This Check Prove?

The command submits the updated Python file with the Spark Kafka connector. A successful run confirms that the new paths and schema did not interrupt the existing broker connection.

You will see the six json_value strings again. The stream still reaches the topic, so the next edit can focus on parsing and storage.

Stream No Longer Reaching Kafka?

Check that the configuration block sits above the existing spark definition. Confirm that every quote in EVENT_SCHEMA is present.

If the command still fails, help me debug the new stream configuration.

The parser now needs to expand each event envelope before the sink writes it. The Parquet sink keeps those typed rows available for later batch processing.

  • Find the current query = ( block in jobs/stream_events.py.
  • Replace everything from query = ( through spark.stop() with this typed streaming query:
parsed_events = (
    raw_events
    .selectExpr(
        "CAST(value AS STRING) AS json_value",
        "timestamp AS kafka_timestamp",
    )
    .select(
        F.from_json(F.col("json_value"), EVENT_SCHEMA).alias("event"),
        F.col("kafka_timestamp"),
    )
    .select("event.*", "kafka_timestamp")
)

query = (
    parsed_events.writeStream
    .format("parquet")
    .outputMode("append")
    .option("checkpointLocation", CHECKPOINT_PATH)
    .trigger(availableNow=True)
    .start(BRONZE_PATH)
)

query.awaitTermination()
spark.stop()

How Does the Typed Stream Work?

  • selectExpr converts the binary Kafka value into JSON text. It also preserves the broker timestamp as kafka_timestamp.
  • from_json applies EVENT_SCHEMA to produce a structured event column.
  • select("event.*", "kafka_timestamp") expands the event into six typed output columns.
  • checkpointLocation preserves offset progress so later runs process new Kafka records incrementally.
  • availableNow=True drains every currently available record before the query exits.
  • Save jobs/stream_events.py.
  • Return to the Terminal from earlier.
  • Create the bronze dataset by running the updated stream:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0 jobs/stream_events.py

What Happens During This Run?

Spark reads every currently available event from campus-events. It writes typed rows under output/bronze.

The query also records its Kafka offsets under output/checkpoints/bronze. Future runs can continue from that progress.

  • Switch back to VS Code after the Spark job exits.
  • Expand the output folder in the Explorer sidebar.
  • Confirm that the new bronze folder contains Parquet files.
  • Expand the checkpoints folder.
  • Confirm that checkpoints/bronze contains streaming state.

That is the raw-payload problem solved. Your six Kafka records now exist as typed bronze data with durable offset progress.

Missing Bronze or Checkpoint Files?

Confirm that parsed_events.writeStream uses the Parquet format. Check that BRONZE_PATH and CHECKPOINT_PATH point inside output.

If the job exits without creating both folders, help me debug the typed Spark stream.

✔️ Awesome, I've got everything!

Your typed stream file is saved. The bronze dataset and checkpoint state are ready for the summary job.

ⓧ I'd like to double check the full code

from pathlib import Path

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

PROJECT_ROOT = Path(__file__).resolve().parents[1]
BRONZE_PATH = str(PROJECT_ROOT / "output" / "bronze")
CHECKPOINT_PATH = str(PROJECT_ROOT / "output" / "checkpoints" / "bronze")
EVENT_SCHEMA = "event_id STRING, event_time STRING, zone STRING, event_type STRING, count INT"

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("CampusEventIngest")
    .getOrCreate()
)

raw_events = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "campus-events")
    .option("startingOffsets", "earliest")
    .load()
)

parsed_events = (
    raw_events
    .selectExpr(
        "CAST(value AS STRING) AS json_value",
        "timestamp AS kafka_timestamp",
    )
    .select(
        F.from_json(F.col("json_value"), EVENT_SCHEMA).alias("event"),
        F.col("kafka_timestamp"),
    )
    .select("event.*", "kafka_timestamp")
)

query = (
    parsed_events.writeStream
    .format("parquet")
    .outputMode("append")
    .option("checkpointLocation", CHECKPOINT_PATH)
    .trigger(availableNow=True)
    .start(BRONZE_PATH)
)

query.awaitTermination()
spark.stop()

What Should the Full File Contain?

This reference combines the project paths, typed event schema, Kafka source, JSON parser, Parquet sink, checkpoint, and bounded query. Compare it with your saved jobs/stream_events.py.

Build the activity summary

The bronze dataset preserves event-level detail. The summary job groups those rows by zone and event_type. It then adds the typed count values into activity_count.

  • Select the jobs folder in the VS Code Explorer sidebar.
  • Click the New File button.
  • Enter build_summary.py as the file name.
  • Populate jobs/build_summary.py by pasting this summary job:
from pathlib import Path

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

PROJECT_ROOT = Path(__file__).resolve().parents[1]
BRONZE_PATH = str(PROJECT_ROOT / "output" / "bronze")
GOLD_PATH = str(PROJECT_ROOT / "output" / "gold")

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("CampusActivitySummary")
    .getOrCreate()
)

events = spark.read.format("parquet").load(BRONZE_PATH)

summary = (
    events
    .groupBy("zone", "event_type")
    .agg(F.sum("count").alias("activity_count"))
    .orderBy("zone", "event_type")
)

summary.show(truncate=False)
summary.coalesce(1).write.mode("overwrite").json(GOLD_PATH)
spark.stop()

How Does the Summary Job Work?

  • events loads the typed Parquet rows from output/bronze.
  • groupBy creates one group for each zone and event type combination.
  • F.sum("count") calculates the total activity within each group.
  • summary.show prints the grouped table for inspection.
  • coalesce(1) produces one JSON data partition under output/gold.
  • Save jobs/build_summary.py.
  • Return to the Terminal from earlier.
  • Build the first gold summary by running this command:
spark-submit jobs/build_summary.py

What Does This Run Create?

The command loads the bronze Parquet files as a batch DataFrame. It prints the grouped activity table before writing the same result under output/gold.

You will see columns named zone, event_type, and activity_count. The output/gold folder now contains one JSON data partition.

Summary Cannot Read the Bronze Data?

Confirm that output/bronze contains Parquet files from the streaming job. Check that BRONZE_PATH uses the same path in both Python files.

If the summary still fails, help me debug the bronze-to-gold Spark job.

✔️ Awesome, I've got everything!

Your summary job is saved. It reads the bronze dataset and produces the gold JSON data product.

ⓧ I'd like to double check the full code

from pathlib import Path

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

PROJECT_ROOT = Path(__file__).resolve().parents[1]
BRONZE_PATH = str(PROJECT_ROOT / "output" / "bronze")
GOLD_PATH = str(PROJECT_ROOT / "output" / "gold")

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("CampusActivitySummary")
    .getOrCreate()
)

events = spark.read.format("parquet").load(BRONZE_PATH)

summary = (
    events
    .groupBy("zone", "event_type")
    .agg(F.sum("count").alias("activity_count"))
    .orderBy("zone", "event_type")
)

summary.show(truncate=False)
summary.coalesce(1).write.mode("overwrite").json(GOLD_PATH)
spark.stop()

What Should the Full File Contain?

This reference contains the bronze reader, grouped aggregation, terminal table, JSON writer, and Spark shutdown. Compare it with your saved jobs/build_summary.py.

Verify the complete data product

A complete data product needs repeatable processing plus an inspectable result. This final check runs the ingestion path before rebuilding the summary from the typed bronze data.

Before you run the final check, which totals do you expect for library entry activity and north shuttle arrival activity?

  • Run the complete transformation path and print the gold JSON by using these commands:
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0 jobs/stream_events.py
spark-submit jobs/build_summary.py
cat output/gold/part-*.json

What Does the Final Check Prove?

The checkpoint lets the ingestion query finish without duplicating previously processed Kafka offsets. The batch job rebuilds the gold summary from the current bronze dataset.

The final command prints the JSON records stored in the gold output partition. This gives you a terminal-visible result that can be checked without another application.

You will see the typed summary table before the JSON records. The library and entry record has an activity_count of 2. The north_shuttle_stop and arrival record has an activity_count of 14.

You now have a typed activity data product. Kafka events flow into checkpointed bronze Parquet before Spark turns them into a queryable gold summary.

Totals Missing or Incorrect?

Confirm that data/events_batch_1.jsonl was published once to campus-events. Check that the summary groups by both zone and event_type.

If the JSON output differs from the printed table, help me trace the Spark aggregation output.

Your typed bronze layer and gold activity summary are ready. Next, Apache Airflow turns these two Spark jobs into one ordered workflow that you can trigger and monitor.

Orchestrate Incremental Processing

Your Kafka events now land as typed Parquet records. Your Spark summary already turns those records into a queryable JSON result.

Manual launches leave task order in your hands. Failures can hide in separate Terminal output.

Apache Airflow makes the two-stage workflow explicit. Its DAG runs ingestion before the summary rebuild.

In this step, get ready to:
  • Define an Airflow DAG with two ordered Spark tasks.
  • Publish four new campus events before starting Airflow.
  • Trigger the DAG to verify incremental processing in the Airflow UI.
Create the Airflow DAG

An Airflow DAG records workflow tasks as code. A dependency between the tasks ensures that the summary only rebuilds after ingestion succeeds.

  • Create dags/campus_activity_pipeline.py from the VS Code Explorer.
  • Add the DAG configuration plus the ingestion task by pasting this code:
from pathlib import Path

import pendulum
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG

PROJECT_ROOT = Path(__file__).resolve().parents[1]
SPARK_KAFKA_PACKAGE = "org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0"

with DAG(
    dag_id="campus_activity_pipeline",
    schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["data-engineering", "campus"],
) as dag:
    stream_events = BashOperator(
        task_id="stream_events_to_bronze",
        bash_command=(
            "set -e; spark-submit --packages "
            f"{SPARK_KAFKA_PACKAGE} jobs/stream_events.py"
        ),
        cwd=str(PROJECT_ROOT),
    )

What does this code do?

  • The PROJECT_ROOT path gives each task the campus-pulse directory as its execution location.
  • The SPARK_KAFKA_PACKAGE value supplies the Kafka connector required by the streaming job.
  • The schedule=None setting keeps this workflow manually triggered.
  • The first BashOperator submits the checkpointed Kafka ingestion job.
  • Save dags/campus_activity_pipeline.py.
  • Confirm that dags/campus_activity_pipeline.py appears inside the dags directory in the Explorer.

Cannot find the DAG file?

Confirm that the file sits inside campus-pulse/dags. A file created beside the campus-pulse directory stays outside Airflow's configured DAG folder.

If the location still looks wrong, help me check my Airflow DAG path.

  • Place your cursor below the closing parenthesis for stream_events.
  • Add the summary task plus its dependency by pasting this code:
    build_summary = BashOperator(
        task_id="build_campus_activity_summary",
        bash_command="set -e; spark-submit jobs/build_summary.py",
        cwd=str(PROJECT_ROOT),
    )

    stream_events >> build_summary

How is the workflow ordered?

  • The build_summary task submits the existing batch summary job.
  • The stream_events >> build_summary dependency makes ingestion finish before aggregation starts.
  • The set -e; prefix propagates a failed Spark command back to Airflow.
  • Save dags/campus_activity_pipeline.py.
  • Confirm that the final line reads stream_events >> build_summary.

Seeing a Python warning in VS Code?

Check that both task definitions remain indented inside the with DAG block. The dependency line needs the same indentation as the task definitions.

If the indentation still looks unclear, help me compare my Airflow DAG structure.

✔️ Awesome, I've got everything!

Your DAG now contains the ingestion task plus the summary task. The dependency places them in the correct execution order.

ⓧ I'd like to double check the full code

from pathlib import Path

import pendulum
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG

PROJECT_ROOT = Path(__file__).resolve().parents[1]
SPARK_KAFKA_PACKAGE = "org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0"

with DAG(
    dag_id="campus_activity_pipeline",
    schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["data-engineering", "campus"],
) as dag:
    stream_events = BashOperator(
        task_id="stream_events_to_bronze",
        bash_command=(
            "set -e; spark-submit --packages "
            f"{SPARK_KAFKA_PACKAGE} jobs/stream_events.py"
        ),
        cwd=str(PROJECT_ROOT),
    )

    build_summary = BashOperator(
        task_id="build_campus_activity_summary",
        bash_command="set -e; spark-submit jobs/build_summary.py",
        cwd=str(PROJECT_ROOT),
    )

    stream_events >> build_summary

How should I compare the file?

Compare the imports first. Next, compare the two task IDs plus the final dependency line.

Publish the next batch and start Airflow

The existing checkpoint already records the first six Kafka offsets. A second event batch gives the DAG new data to process during its first run.

  • Create data/events_batch_2.jsonl from the VS Code Explorer.
  • Add four new campus events by pasting this data:
{"event_id":"evt-007","event_time":"2026-09-03T08:03:00Z","zone":"library","event_type":"checkout","count":3}
{"event_id":"evt-008","event_time":"2026-09-03T08:03:30Z","zone":"north_shuttle_stop","event_type":"arrival","count":11}
{"event_id":"evt-009","event_time":"2026-09-03T08:04:00Z","zone":"gym","event_type":"exit","count":1}
{"event_id":"evt-010","event_time":"2026-09-03T08:04:30Z","zone":"library","event_type":"entry","count":2}

What is in this batch?

This batch adds four later events with IDs from evt-007 through evt-010. Their timestamps follow the first batch.

The new library entry count is 2. The new north shuttle arrival count is 11.

  • Save data/events_batch_2.jsonl.
  • Confirm that the file contains exactly four lines.

Does the file show JSON errors?

Keep each event on one line. Check that every property name uses double quotes.

If VS Code still highlights the records, help me validate my JSON Lines file.

✔️ Awesome, I've got everything!

Your second event batch is saved with four complete records.

ⓧ I'd like to double check the full code

{"event_id":"evt-007","event_time":"2026-09-03T08:03:00Z","zone":"library","event_type":"checkout","count":3}
{"event_id":"evt-008","event_time":"2026-09-03T08:03:30Z","zone":"north_shuttle_stop","event_type":"arrival","count":11}
{"event_id":"evt-009","event_time":"2026-09-03T08:04:00Z","zone":"gym","event_type":"exit","count":1}
{"event_id":"evt-010","event_time":"2026-09-03T08:04:30Z","zone":"library","event_type":"entry","count":2}

How should I compare the batch?

Compare each event ID plus its count. The file should end with evt-010.

  • Publish all four records to campus-events by running this command in the activated Terminal:
docker compose exec -T kafka /opt/kafka/bin/kafka-console-producer.sh --topic campus-events --bootstrap-server localhost:9092 < data/events_batch_2.jsonl

What does this command do?

Docker Compose runs the Kafka producer inside the existing kafka service. The input redirection sends each line from data/events_batch_2.jsonl to the campus-events topic.

The -T option disables pseudo-terminal allocation. This allows the file stream to reach the producer unchanged.

You will return to the Terminal prompt after the producer accepts the file. The broker now holds ten campus events.

Did the producer command fail?

Confirm that Docker Desktop still has the Kafka service running. Check that data/events_batch_2.jsonl exists inside campus-pulse.

If the command still fails, help me debug the Kafka producer command.

Airflow needs a metadata location plus the exact directory that contains your DAG. Standalone mode initializes the local database before serving the web interface.

  • Keep the current activated Terminal inside the campus-pulse directory.
  • Export the Airflow paths plus start standalone mode by running these commands:
export AIRFLOW_HOME="$PWD/.airflow"
export AIRFLOW__CORE__DAGS_FOLDER="$PWD/dags"
airflow standalone

What happens during startup?

  • The AIRFLOW_HOME export places local metadata under .airflow.
  • The AIRFLOW__CORE__DAGS_FOLDER export points DAG discovery at the project's dags directory.
  • The airflow standalone command initializes the database before starting Airflow's local components.
  • Leave this Terminal session running.
  • Open http://localhost:8080 in your browser.
  • Use the locally generated credentials from the Terminal if Airflow asks you to sign in.
  • Keep the generated password out of screenshots.

The first startup can take a few minutes while Airflow initializes its metadata. You will see the Airflow interface once the local services are ready.

Cannot reach the Airflow UI?

Keep the standalone Terminal open. Check its latest log lines for a startup failure before retrying the browser.

If startup stopped unexpectedly, help me debug Airflow standalone.

Trigger and verify the incremental workflow

The DAG is now available to Airflow. Its first task uses the existing checkpoint to find newly available Kafka offsets.

  • Create a second Terminal session from the Terminal panel.
  • Activate .venv in the new session using the same activation command from earlier.

Before you trigger the DAG, do you think the checkpoint makes Spark process all ten events or only the four new events?

  • Connect the second Terminal to the same Airflow metadata before triggering the workflow by running these commands:
export AIRFLOW_HOME="$PWD/.airflow"
export AIRFLOW__CORE__DAGS_FOLDER="$PWD/dags"
airflow dags unpause campus_activity_pipeline
airflow dags trigger campus_activity_pipeline

What do these commands do?

  • The exports connect this Terminal to the same Airflow metadata plus DAG directory.
  • The airflow dags unpause command allows the workflow to run.
  • The airflow dags trigger command creates a manual run for campus_activity_pipeline.

The first orchestrated run may stay active for a few minutes while Spark resolves the Kafka connector. The run is still progressing while either task shows a running state.

  • Return to http://localhost:8080.
  • Select campus_activity_pipeline from the DAG list.
  • Open the latest triggered run.
  • Inspect the task graph until both tasks are green.
  • Inspect the ingestion task log to confirm that the Spark job completed.

You will see stream_events_to_bronze complete before build_campus_activity_summary. Both tasks finish in green.

Before you inspect the refreshed output, what totals do you expect after adding the second batch to the existing bronze data?

  • Read the refreshed gold JSON output by running this command:
cat output/gold/part-*.json

What does this check prove?

The file pattern reads the single JSON partition written by jobs/build_summary.py. Its totals come from every typed record currently stored in bronze.

The checkpoint prevented the first six Kafka records from being appended to bronze again. Only evt-007 through evt-010 were newly ingested.

You will see library entry activity of 4 plus north shuttle arrival activity of 25. These totals confirm that the checkpoint advanced through evt-010.

Are the totals unchanged?

Confirm that both Airflow tasks finished in green. Check the ingestion task log if the summary task never started.

If both tasks succeeded but the totals still show the first batch, help me trace my incremental Airflow run.

That is the hard part done. Your local pipeline now ingests only new Kafka events before rebuilding the campus activity summary through one monitored Airflow run.

Secret mission

Quarantine Unknown Campus Zones

Your activity summary currently trusts every campus zone that reaches bronze storage. Add a quality gate that keeps approved zones in the gold summary while routing an unexpected roof garden event into a separate quarantine dataset.

Clean Up Your Resources

Clean Up Your Resources

Campus Pulse runs entirely on your Mac, so leaving its files in place costs $0. Choose whether to keep the services running, pause them for later, or delete the local project.

Resources you used:

  • Local broker: One Apache Kafka container plus its Docker Compose network. The container holds the campus-events topic.
  • Local orchestration metadata: The .airflow directory containing the Apache Airflow database plus successful DAG run history.
  • Generated pipeline data: The output directory containing Apache Spark checkpoints plus bronze Parquet data. It also contains gold and quarantine JSON outputs.
  • Project environment and source files: .venv, data, jobs, dags, and compose.yaml inside the campus-pulse folder.

Keep everything running

No action needed. Choose this if you want to publish more events or inspect another Airflow run.

  • Keep the current Airflow process running in its Terminal session.
  • Keep the Kafka service running through Docker Compose.
  • Keep the campus-pulse folder on your Desktop so its metadata, checkpoints, datasets, virtual environment, and source files remain available.
  • Use http://localhost:8080 whenever you want to inspect the existing DAG runs.

Pause - I'll come back to this later

Stop the running processes to free local memory. Your Kafka topic, Airflow history, checkpoints, datasets, and source files remain intact.

  • Press Control-C in the Airflow Terminal to stop the current standalone process.
  • Stop the Kafka container without removing it by running this command:
docker compose stop

What Happens When Kafka Pauses?

This command stops the Kafka container without removing the container or its Compose network. The campus-events topic remains available when you restart the same container.

Does the Pause Command Fail?

  • Confirm Docker Desktop is running.
  • Return to the activated Terminal session in campus-pulse.
  • Ask for help with the pause command.

When you return, restart the broker before Airflow so the ingestion task can reach Kafka.

  • Return to the campus-pulse folder in an activated Terminal.
  • Restart the Kafka container by running this command:
docker compose up -d

What Happens When Kafka Restarts?

Detached mode starts the broker in the background. Your Terminal remains available for the next command.

Does the Broker Stay Offline?

  • Confirm Docker Desktop is running.
  • Confirm compose.yaml is inside the current campus-pulse folder.
  • Ask for help restarting the broker.
  • Restore the same AIRFLOW_HOME export from earlier.
  • Restore the same AIRFLOW__CORE__DAGS_FOLDER export from earlier.
  • Start Airflow by running this command:
airflow standalone

What Happens When Airflow Restarts?

Airflow starts its local components from the existing metadata directory. This Terminal remains occupied until you stop the process with Control-C.

Does Airflow Fail to Start?

  • Reactivate .venv using the same method from Step 1.
  • Restore both Airflow environment exports from earlier.
  • Ask for help restarting Airflow.
  • Open http://localhost:8080 in your browser to confirm the Airflow UI is available again.

You should see the existing campus_activity_pipeline run history, including the successful quality-gate run.

Delete - I don't want to use this again

Deleting the local folder can feel drastic. This option removes every Campus Pulse artifact while leaving the Homebrew Spark installation available.

  • Copy any event files or DAG code that you want to keep.
  • Press Control-C in the Airflow Terminal to stop the current standalone process.
  • Return to the activated Terminal session in campus-pulse.
  • Remove the Kafka container plus its Compose network by running this command:
docker compose down

What Does This Remove?

This command stops and removes the Kafka container plus the Compose network. Removing the container also removes its local campus-events topic data.

Does the Teardown Command Fail?

  • Confirm Docker Desktop is running.
  • Confirm the current Terminal is inside campus-pulse.
  • Ask for help removing the Compose resources.
  • Use macOS search to open your file browser.
  • Delete the campus-pulse folder from your Desktop.
  • Empty the deleted-items folder on your Mac.
  • Check your Desktop for the campus-pulse folder.

You should no longer see campus-pulse. Removing that folder also removes .airflow, output, .venv, and every project source file.

Nice Work!

Nice Work!

Mission complete. Your Campus Pulse pipeline now turns live Kafka events into an Airflow-orchestrated data product through Spark Structured Streaming.

You've learned how to:

  • Publish repeatable JSON campus events to a Kafka topic. Read each record back to prove the broker provides a durable handoff.
  • Transform opaque Kafka payloads into typed columns with Spark Structured Streaming. Write a bronze Parquet dataset with checkpointed offsets for incremental processing.
  • Produce a queryable JSON activity summary grouped by campus zone and event type. Orchestrate the two Spark jobs with an Airflow DAG. Confirm both tasks succeed in the Airflow UI.
  • Complete the optional Secret Mission by adding a quality gate for unknown campus zones. Keep the roof_garden event out of trusted totals. Preserve it in the quarantine dataset.

Ready to quiz yourself?