Missing spark structured streaming metadata blobs?

[DB1To3]

I'm exploring spark structured streaming, and contemplating uses for it.

I'd like to launch a job on a recurring basis from an external scheduler.  It will launch a spark job with trigger (availableNow=True).

I'm concerned about overlapping executions, and corruption of the "checkpointLocation".  As I was reading the docs, it seems like a folder is needed for that:_spark_metadata.  It has benefits like:

• Protect against concurrent processing of the same query to the same target
• Ensure "exactly once" processing of inputs

However whenever I search for that folder (_spark_metadata) in blob storage I'm unable to find it.  I suspect this has something to do with "toTable" that delivers data into a UC managed-table.

Is there any way I can deliver data to the UC managed-table without losing the benefits of the _spark_metadata folder?  I'd like to have a consistent set of code that can be written to deliver data to a flexible target, without losing the _spark_metadata folder in the process.  I don't see why the delivery of streaming to a table is mutually excluding the ability to generate the _spark_metadata blobs.  Any tips would be appreciated.

[Yougasundar]

Hi @DB1To3,

You do not lose anything here. The _spark_metadata folder belongs to the Spark file sink (for example format("parquet").start(path)), so it does not appear when you write with toTable to a Delta table. For a Delta sink, the Delta transaction log does that job. The Databricks docs say it "guarantees exactly-once processing, even when there are other streams or batch queries running concurrently against the table." Your checkpointLocation still keeps the offsets and commits of each micro-batch, so the next availableNow run continues where the last run stopped. If one of your flexible targets is a plain file path, Spark creates _spark_metadata there by itself, so the same code works for both target types. The docs also say that each query must have its own checkpoint, and that queries must never share one. So the real risk is two runs of the same query at the same time, not the missing folder. A simple guard: put the code in a Lakeflow Job and let your external scheduler call the Jobs API "run now". New jobs have "maximum concurrent runs" set to 1, so an overlapping trigger is queued or skipped and never runs in parallel.

Docs: https://docs.databricks.com/aws/en/structured-streaming/delta-lake

[DB1To3]

Thanks for the thorough answer.
When we use file/blob output, we use  .option("path", xyz) and that is the place where the _spark_metadata lives.  If we stream to a table then we don't use that option anymore - and we lose the folder. My concern was that when we switch to tables we lose something we otherwise had when streaming to files.

>> So the real risk is two runs of the same query at the same time, not the missing folder. 

Exactly.  Much of my understanding is from LLM responses so take with a grain of salt, but I heard the _spark_metadata manages concurrency and allows only one writeStream to be delivering data to that query-and-sink at any time.  It doesn't seem to be very well documented, possibly because it is a feature of the sink instead of a feature of spark structured streaming as a whole.

It continues to baffle me that there is no built-in concurrency control for something so fundamental as this (preventing the writing of the same query to the same sink by concurrent drivers).  I suppose the Lakeflow Job approach works too, but it seems a bit overkill.  A blob-lease seems better, but I haven't found the discussions about doing it that way (yet).  Possibly because that is what _spark_metadata is doing internally?  I'm a bit new to databricks; does everyone use lakeflow jobs for critical sections?  I appreciate the creative approach, but it does not seem like a versatile way to solve the generalized need for one thing happening at a time in this platform.  Maybe everyone will start using lakebase tables for that?    

[Yougasundar]

Hi @DB1To3,

Good follow-up, and you are right that you lose nothing important. The _spark_metadata folder is the file sink's own commit log: it records which files each batch committed, so that readers of that path skip files from failed or repeated batches. It is not a lock. With a table sink, the Delta transaction log does that commit job, which is why the folder is not needed there. The concurrency guard you are looking for lives in the checkpoint, and it works the same for file and table sinks. The Databricks error reference lists two errors for this case: CONCURRENT_QUERY ("Another instance of this query ... was just started by a concurrent session") and CONCURRENT_STREAM_LOG_UPDATE ("Concurrent update to the log. Multiple streaming jobs detected for <batchId>. Please make sure only one streaming job runs on a specific checkpoint location at a time."). So if two drivers start the same query on the same checkpoint, the second one is expected to fail with one of these errors instead of silently writing the batch again. This is fail-fast detection, not a queue, so your scheduler should either avoid the overlap or treat that error as "already running". That is why the Lakeflow Job setting is common for scheduled streams: with maximum concurrent runs set to 1, an overlapping trigger is queued or skipped before a second driver starts, and your external scheduler can still start it with the Jobs API "run now". A blob lease or a lock row in a Lakebase table (it is Postgres, so normal row locks work) is fine for a general "one at a time" need, but for streaming the checkpoint already gives you the protection that the _spark_metadata folder never did.

Docs: https://docs.databricks.com/aws/en/error-messages/error-classes

[DB1To3]

Thanks.  I will do some testing and see if I can get those concurrency error messages to appear.  It is one of the times when you really, really want to see those errors, because things would be a lot worse without them!

If you know of any blogs or docs that I'm overlooking, please let me know.  Your simple statement, "concurrency guard you are looking for lives in the checkpoint", is not something I found yet in my reading.

[saurabh18cs]

Hi @DB1To3 Just make this concurrent run to 1 for your streaming jobs are your problem is sorted , it wont corrupt checkpoint locatiion

[DB1To3]

What if I'm not using lakeflow jobs?  Eg. what if I'm using the REST api to submit jobs (POST /api/2.2/jobs/runs/submit to create ephemeral runs)

Are you also confirming that spark structured streaming won't protect me from duplicate/overlapping executions of the same query-and-sink?  It is odd that I don't find more discussions about this.  It is hard to google, especially prior to experiencing the symptoms of the concurrency conflicts (and potential corruption).