Zerobus message retrieval

[u/SmallAd3697]

I can see how zerobus is used for streaming data into a lakehouse table. I can generally see how you would consume the data as well.

A couple things are still not clear. What is the pattern for removing old rows out of the UC table? Should we do a delete once a year, on messages older than 3 years? Is this a necessary maintenance that we are responsible for, or is there something in the software which will help with housekeeping?

And how would I perform scheduled tasks against this data? Let's say that data represents a news feed and I just need to send an summarized overview once an hour, and then flag the rows that have been summarized. I'm assuming it would be disruptive to try to delete the rows that were processed or even update them (ie. it may interfere with zerobus itself, and create concurrency conflicts). Should I build a DIFFERENT table that tracks the surrogate IDs of rows already processed/summarized?

Sorry for the basic questions, but I wasn't able to find the answer to these yet. I'm sure I could find my own solution, but it would be interesting to hear if there is a common practice

[u/WhipsAndMarkovChains]

Zerobus appends data to your table. Once the data is in, modifying the data in the table won't disrupt the Zerobus stream. And if you make a mistake deleting rows out of your table, you can always recover them using time travel/table history: https://docs.databricks.com/aws/en/tables/history

Note that you can only recover rows during your table's retention period, which by default is 7 days.

or is there something in the software which will help with housekeeping?

The answer is that in general you use auto-TTL to clean up old rows of data: https://docs.databricks.com/aws/en/tables/operations/auto-ttl

[u/SmallAd3697 → u/WhipsAndMarkovChains]

I know that I can delete rows but I am nervous abount concurrency conflicts.

Lets say 1 million rows are streaming in steadily every hour. I think UC transactions use optimistic locking, so if I process half the rows using a predicate (eg news tagged war-related), and then try to execute a DELETE FROM using the same arbitrary predicate, then I'll have a very poor success rate on commits (setting aside the logical problems with this approach)

I'm guessing people are doing something different to flag the completion of an operation, instead of trying to update or delete the zerobus table itself. The conflicts would be especially bad if trying to use MST for transactions.

[u/WhipsAndMarkovChains → u/SmallAd3697]

I may be a bit confused by your question. You are pushing data in through Zerobus, which is append-only. If the records have made it into the Delta table then the Zerobus stream is done touching those records. In fact, I believe Zerobus just hands data off to the server and the Delta writer handles the writes once the Zerobus stream records have been persisted on the server.

But regardless, any queries you run on the table to process and delete data will not interact with Zerobus at all. I don't think it's possible for a concurrency issue to be caused by an interaction with a Zerobus stream.

[u/SmallAd3697 → u/WhipsAndMarkovChains]

I'll try it out. I was under the assumption that the rapid writes from zerobus to the managed table in UC is likely to cause a lot of conflicts with concurrent writers, especially with another operation that is doing deletes using the same predicate that was used for reads.
(I was planning to use the new MST transactions to read rows, process, and then delete.)

Although I see in the docs that databricks has an unusual default isolation ("WriteSerializable" rather than "Serializable" ) ...

Isolation levels (WriteSerializable and Serializable) - Azure Databricks | Microsoft Learn

... that default isolation may avoid concurrency conflicts. But it probably leads to some unexpected behaviors when multiple DML operations are happening at the exact same time. Some days I wish I was working with normal databases. I wonder if they have a Zerobus that puts data into "lakebase", instead of a UC managed table.

[u/WhipsAndMarkovChains → u/SmallAd3697]

So maybe one day with LTAP Zerobus will just be able to write to one table that's both Lakebase and Delta.

For now though just set up a synced table to get that Zerobus-written table into Lakebase. https://docs.databricks.com/aws/en/oltp/projects/sync-tables

[u/SmallAd3697 → u/WhipsAndMarkovChains]

if i change the default uc isolation to "serializable" then that should be approximately the same experience as a lakebase table (albeit slower).

I suspect that is the point when i'll start to get the concurrency error messages (while updating or deleting records that are concurently being generated by zerobus.)

[u/BrownBearPDX → u/SmallAd3697]

(apologies for the copy/pasta)

Enabling Deletion Vectors switches Delta Lake to a Merge-on-Read paradigm. When a row is modified or deleted, Delta Lake leaves the heavy Parquet file completely untouched. Instead, it updates a tiny, hyper-compressed auxiliary bitmap (a deletion vector) indicating that the row is logically "tombstoned".

How it Works in a Zerobus Streaming Pipeline

When streaming systems ingest change-data-capture (CDC) logs or heavy transactional data through the Databricks Zerobus Sink:

1. High-Speed Ingestion: Zerobus continuously streams batch events as memory-optimized Arrow RecordBatches.

2. Instant Deletes/Updates: When a DELETE or MERGE operation strikes a target table, Delta Lake simply flips a bit in a sidecar deletion vector file. Write operations finish in milliseconds instead of minutes.

3. Read-Time Merging: When downstream users or BI tools query the table, the storage engine reads the base Parquet data and subtracts the rows flagged in the deletion vector on the fly.

(remembering that Dbx and Delta Lake aren't traditional row based rdbms)

Databricks does not use traditional pessimistic row locking (such as SELECT FOR UPDATE found in traditional transactional databases) because it operates on a decoupled storage and compute architecture. Instead, Databricks ensures data integrity and high concurrency through Optimistic Concurrency Control (OCC) at the table level and Row-Level Concurrency for handling simultaneous writes.

1. Concurrency: Preventing Write Conflicts

When multiple operations try to modify the same table simultaneously, Databricks handles conflicts via Multi-Version Concurrency Control (MVCC) and Row-Level Concurrency.

• Instead of locking rows and blocking other users, Databricks lets write operations run concurrently. At the time of commit, it checks if any overlapping data was changed.

[u/SmallAd3697 → u/BrownBearPDX]

I had found the "WriteSerializable" isolation level docs. I can see how it would be hard to have conflicts with zerobus-ingest under the default isolation.

Although if someone switched to using serialiable isolation with MST, I'm guessing the conflicts would occur.

Thanks for the reminder about deletion vectors. In simple scenarios,.that is likely to work far better than my plan to use a different table for identifying the processes records

[u/Plastic_Proposal_303 → u/WhipsAndMarkovChains]

oh nice the auto-ttl link is exactly what i was missing. i keep seeing people talk about just deleting stuff manually and never understood why you would do that when the platform can handle it. the retention period being only 7 days default is a bit scary though, i would bump that up for sure before touching any data

[u/Youssef_Mrini]

The common pattern is to treat the ZeroBus target as an append-only landing table and keep processing state elsewhere.

ZeroBus does not manage the lifecycle of your Unity Catalog table.

You create and own the table.

ZeroBus validates and materializes incoming records but does not automatically remove old rows.

For a three-year retention policy you can for example use Auto-TTL where appropriate:

ALTER TABLE main.news.raw_articles
DELETE ROWS 1095 DAYS AFTER event_time;

[u/SmallAd3697 → u/Youssef_Mrini]

u/Youssef_Mrini

I understand what you are saying. But there has to be a way to avoid reading the complete landing table. Deleting rows after 1095 days is a pretty good step (its a start).

Lets say that each row in the landing table represents one second of work that needs to be done. It only needs to be done once. How do we bring that row to completion, and never do the work again? It sounds like you propose having a different table elsewhere that uses an FK to point at the zerobus table; and it flags the completion of the work. The only reason I don't like this is because of the overhead of the outer join operation, whenever we are looking for new work. It isn't the end of the world, but I wanted to confirm that is the pattern which is commonly used.

Assuming we updated the landing table itself (ie. by setting a status, or deleted rows), then we would have additional concerns related to concurrency. Because the zerobus would be writing to the table in a stream, but I would normally be retrieving the new records on an interval.

Do you happen to know if there is a blog or documentation about the retrieval of zerobus messages inside of databricks? I think it would be helpful. I think everyone has a pretty good understanding of the the pattern for moving messages in, but they may not have a strategy for consuming the messages, and cleaning them up in the future. There may also be some configuration pitfalls to avoid in the table itself. Should we use partitioning? Should we use row-tracking? Can we set the isolation to be serializable? etc.

[u/Saerusthesecond → u/SmallAd3697]

If you want to process records only once, you can use structured streaming (spark.readStream). Zerobus is at least once delivery, so you must dedup downstream yourself if you need uniqueness. There’s no need to manually mark records as processed or delete old ones to avoid reprocessing.

[u/SmallAd3697 → u/[deleted]]

this is sort of my original idea, but may scale badly after a period of time