- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
2 weeks ago
Streaming tables are append-only by design, and a materialized view recomputes over everything in the folder. Neither gives you "replace the table with just the newest file" out of the box, so you need a small pattern on top. Here are the two I'd consider.
Option 1: Auto Loader + foreachBatch overwrite (my preference)
Auto Loader tracks which files it has already processed and only picks up new ones on each run [1]. Instead of appending, you overwrite the target table with whatever each batch contains. That means the table always holds only the newly arrived file.
def overwrite_orders(batch_df, batch_id):
if batch_df.isEmpty():
return # nothing new, leave the table untouched
(batch_df.write
.mode("overwrite")
.saveAsTable("my_catalog.my_schema.orders"))
(spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("header", "true")
.option("cloudFiles.schemaLocation", "/Volumes/my_catalog/my_schema/chk/orders_schema")
.load("/Volumes/my_catalog/my_schema/landing/orders/")
.writeStream
.foreachBatch(overwrite_orders)
.option("checkpointLocation", "/Volumes/my_catalog/my_schema/chk/orders")
.trigger(availableNow=True)
.start())
Notes:
- Run this as a normal job or notebook task, scheduled daily or file-arrival triggered. Don't try to put it inside a declarative pipeline streaming table, because that is the append model you're already seeing.
- The
isEmpty()guard matters. Without it, a run with no new files would overwrite your table with an empty result. - If two files land between runs, one batch will contain both. If you need strictly "only the latest file", filter the batch on
_metadata.file_modification_time(or_metadata.file_path) before writing. - Delta overwrite is atomic, so readers never see a half-empty table, and you keep time travel on the previous version.
Option 2: Materialized view that selects only the latest file
If you want to stay fully declarative, keep the materialized view but filter to the newest file using the file metadata column:
CREATE OR REFRESH MATERIALIZED VIEW orders AS
WITH src AS (
SELECT *, _metadata.file_modification_time AS file_ts
FROM read_files(
'/Volumes/my_catalog/my_schema/landing/orders/',
format => 'csv',
header => true
)
)
SELECT * EXCEPT (file_ts)
FROM src
WHERE file_ts = (SELECT max(file_ts) FROM src);
The trade-off is that the source folder is still scanned on refresh, so as old files pile up the cost grows. If you go this route, archive or delete processed files periodically.
What I'd avoid
TRUNCATEfollowed byCOPY INTO.COPY INTOis idempotent and skips files it has already loaded [1], so it would load just the new file. But the two steps aren't atomic, and a run with no new file leaves you with an empty table.- Full refresh on the streaming table. It reprocesses everything in the folder, which is the opposite of what you want.
One more thing to check
If "overwrite" actually means that rows in the new file should replace matching rows (same order_id) rather than wipe the table, you don't want an overwrite at all. You want an upsert, either with MERGE INTO inside foreachBatch or with the declarative auto CDC flow in a pipeline. Let me know which semantics you need and I can sketch that version.
References
[1] Ingest data from cloud object storage | Databricks on AWS — https://docs.databricks.com/aws/en/ingestion/cloud-object-storage
Solutions Engineer @ Databricks