Streaming tables vs materialized views: refresh and full refresh

Part 1 built the shop's pipeline: bronze.py, silver.py and gold.sql in the transformations/ folder of a project made by spark-pipelines init --name shop, with make_files.py writing 2 days of order files into landing/. To start here, make those files from part 1, write both days and run the pipeline once with spark-pipelines run --conf spark.sql.catalogImplementation=hive. Every run below uses that same command.

Run it again with nothing new

Run the pipeline again, before any new file arrives. It ends with Run is COMPLETED., and python show.py bronze_orders prints the same 5 rows. The 2 kinds of table did different things:

• bronze_orders, the streaming table, read nothing. Its checkpoint, in the pipeline-storage/ folder named in the spec, lists both files as read.
• orders and customer_spend, the materialized views, were computed again from what they read, and came out the same.

That difference decides what happens when a query changes.

Change a query in a streaming table

Say someone wants statuses in capitals, and adds 1 line to bronze.py, after .load("landing"):

        .withColumn("status", F.upper("status"))

Then the shop's system drops the file of 2026-10-03, which cancels order 1002 and adds order 1005:

printf 'order_id,customer_id,status,amount,updated_at\n1002,C02,cancelled,12.00,2026-10-03 09:30:00\n1005,C02,placed,18.25,2026-10-03 16:20:00\n' > landing/orders-2026-10-03.csv

Run the pipeline, then python show.py bronze_orders customer_spend:

bronze_orders
+--------+-----------+---------+------+-------------------+---------------------------------------+---------------------+
|order_id|customer_id|status   |amount|updated_at         |_bad                                   |source_file          |
+--------+-----------+---------+------+-------------------+---------------------------------------+---------------------+
|1001    |C01        |placed   |25.50 |2026-10-01 09:15:00|NULL                                   |orders-2026-10-01.csv|
|1001    |C01        |shipped  |25.50 |2026-10-02 08:00:00|NULL                                   |orders-2026-10-02.csv|
|1002    |C02        |CANCELLED|12.00 |2026-10-03 09:30:00|NULL                                   |orders-2026-10-03.csv|
|1002    |C02        |placed   |12.00 |2026-10-01 10:02:00|NULL                                   |orders-2026-10-01.csv|
|1003    |C03        |placed   |NULL  |2026-10-01 11:30:00|1003,C03,placed,abc,2026-10-01 11:30:00|orders-2026-10-01.csv|
|1004    |C01        |placed   |40.00 |2026-10-02 12:45:00|NULL                                   |orders-2026-10-02.csv|
|1005    |C02        |PLACED   |18.25 |2026-10-03 16:20:00|NULL                                   |orders-2026-10-03.csv|
+--------+-----------+---------+------+-------------------+---------------------------------------+---------------------+

customer_spend
+-----------+------+-----+
|customer_id|orders|spend|
+-----------+------+-----+
|C01        |2     |65.50|
|C02        |2     |30.25|
+-----------+------+-----+

Only the 2 new lines are in capitals. A streaming table never reads a row twice, so the rows it wrote before the change keep the old query's result. Databricks' docs give the same example with LOWER and UPPER. Gold is now wrong: silver took CANCELLED as 1002's newest state, and status != 'cancelled' does not match it, so C02's cancelled order still counts.

Rebuild with a full refresh

A full refresh clears a table's rows and checkpoint and computes it from the start. Run it for bronze:

spark-pipelines run --conf spark.sql.catalogImplementation=hive --full-refresh bronze_orders

The log says:

2026-10-18 08:15:12: Flow 'spark_catalog.default.orders' is EXCLUDED.
2026-10-18 08:15:12: Flow 'spark_catalog.default.customer_spend' is EXCLUDED.
2026-10-18 08:15:14: Flow spark_catalog.default.bronze_orders has COMPLETED.
2026-10-18 08:15:15: Run is COMPLETED.

Bronze reread all 3 files and every status is now in capitals. But naming 1 table leaves out every other flow, so silver and gold still hold the last run's results until they run again.

Capitals in bronze were the wrong idea anyway. Bronze keeps the data as it arrived, and cleaning belongs in silver, where it is recomputed every run; part 3 does that there. Remove the upper line, and rebuild bronze and update the tables above it in 1 run:

spark-pipelines run --conf spark.sql.catalogImplementation=hive --full-refresh bronze_orders --refresh orders,customer_spend

Bronze is lower case again, 1002 is cancelled in silver, and customer_spend gives C02 1 order and 18.25. --full-refresh-all rebuilds every table, and a run with no option updates every table and reads only new records.

Check before you run

dry-run reads every file and checks the code and the graph, without reading data or writing a table. Change FROM orders in gold.sql to FROM order and run spark-pipelines dry-run --conf spark.sql.catalogImplementation=hive:

2026-10-18 08:15:44: Failed to resolve flow: 'spark_catalog.default.customer_spend'.
Error: [TABLE_OR_VIEW_NOT_FOUND] The table or view `spark_catalog`.`default`.`order` cannot be found. Verify the spelling and correctness of the schema and catalog.

Put orders back. A typo found this way costs a few seconds, not a failed run halfway through.

2 things to know when running locally

• In this course's runs on PySpark 4.2.0, a query on another pipeline table that used withColumn, such as spark.read.table("bronze_orders").withColumn(...), ran, from the second run on, before that table had updated, and read the last run's rows. Write such queries with select, as this module does. withColumn inside bronze's own read, above, was not affected.
• Changing which columns a materialized view has fails on the next run with the local catalog, with an error that the table cannot be altered, and a full refresh does not help. Run DROP TABLE on it in a Spark session with enableHiveSupport(), as show.py makes, then run the pipeline.