Structured Streaming

Need help with alternatives for dropMalformed and Schema Registry URL not supported for Kafka source on a Shared cluster

Hello, We are migrating from Hive metastore to UC and in a phase where we are testing before pushing the upgraded code to production. The issue we are facing is we have Kafka Structured Streaming jobs which are using Dedicated clusters earlier but going forward we will only be using Shared clusters and Serverless eventually. But the Streaming jobs are failing with errors dropMalformed are not supported for Data source Kafka on a Shared cluster kafka.schena.registry.url are not supported for Data source Kafka on a Shared cluster Did anyone face a similar issue and found an alternative to make it work on Shared clusters ? Kindly help us with resolving this. Thankyou submitted by /u/Its__Ram [link] [comments]

Python Asynchronous TransformWithState for Streaming is now GA

Hi folks, we are excited to announce that Python Asynchronous TransformWithState (Async TWS) is now GA for Structured Streaming. What is it? Async TWS speeds up stateful streaming queries in Python by executing the stateful processor concurrently across keys and batching state store operations, which can improve throughput by up to 3x. The async API mirrors the existing synchronous TransformWithState : You implement AsyncStatefulProcessor instead of StatefulProcessor and apply asyncio syntax changes (e.g, async def and await ). Third-party async libraries are welcome but not required to start seeing performance gains. The AsyncStatefulProcessor is a drop-in replacement for the StatefulProcessor so you can switch implementations while reusing your checkpoint. See here for examples. What is supported Python row-based transformWithState Both Micro-Batch and Real-Time Mode DBR 19+ or serverless compute with environment version 6+ Structured Streaming; SDP on the Preview channel, with SDP Classic enabled once it reaches DBR 19 Because handleInputRows and handleExpiredTimer now run concurrently across keys, make sure any shared member state or external calls are safe for concurrent execution. Learn more at https://docs.databricks.com/aws/en/stateful-applications/async Questions and feedback welcome in the comments. submitted by /u/New_Woodpecker_1631 [link] [comments]

Databricks in a SaaS product, and the realtime mode

Hey everyone, i have two questions about databricks and I’d like to hear about your experiences: 1. I have seen a lot of companies who use databricks for their own internal needs and that makes sense. But have you ever seen a SaaS product that uses databricks? For example a products that process customer data to provide some value? Do you know if there is any company that uses databricks structured streaming realtime mode in production? Do you know if they are happy with it? I am asking this because it is a new offering and there is no support for features such as auto scaling (unless if you are using lakeflow pipelines). And i assume it is super expensive as you need a core for every task slot. submitted by /u/shanfamous [link] [comments]

Structured Streaming job on a job cluster ends with "driver unresponsive" after 8-12 h, with a healthy heap and plenty of memory

Setup Azure Databricks, job cluster, Python script task (not a notebook). The JVM runs JDK 21 with ParallelGC and a 101 GB heap, on a driver with 128 GB. 2 workers 28gb 4 cores each. About 7-8 Structured Streaming queries share one driver. They use foreachBatch, and each batch holds roughly 1 record (0.1-0.2 records/s). A batch takes about 6-7 s, with spikes of 12-16 s, and almost all of that time is addBatch. Inside foreachBatch on the driver there is an in-process cache with a TTL. One thread refreshes it under a lock while the others keep serving the old snapshot. Problem After roughly 8-12 hours the run ends. The driver log contains a "DRIVER UNRESPONSIVE STATS" block, and the event log only shows the cluster terminating, with no OOM, resize or eviction. 1 job 8 tasks, all report that driver became unhealthy.. What the metrics show before the crash Driver container memory grows slowly from about 45 to 58 GB, with a 128 GB limit. JVM heap shows a normal sawtooth with no trend. In the dumps, old gen is at 10-19% and metaspace at about 450 MB. Worker-side GC events rise from about 5 to about 70 per interval, and GC pause time from about 0.2 s to 1.5 s per interval. The GC count is high just before the crash, even though memory is nowhere near its limit. What the driver dumps show In one dump, three threads are BLOCKED in ClassLoader.loadClass. They wait on monitors of Databricks' LibraryClassLoader and DriverCorralClassLoader, and one comes in via Janino (findIClass → Class.forName). The thread holding the lock is not in the output, and I can't tell how long they had been waiting. The dump from another crash had no blocked threads at all. Other observations The log shows 16 separate Python processes connecting through Py4J at startup, and another 16 about 30 s later. I don't know why. The same logic ran for more than 72 h on serverless without problems. As far as I know, serverless can't use a continuous trigger, so each micro-batch is a fresh start there. That is not a like-for-like comparison. Questions Is class-loader lock contention on the driver with many concurrent streams a known problem? Could it make the driver miss its health check? What else causes "driver up but not responsive" when heap and metaspace look fine? Is there guidance on how many streaming queries one driver can handle? Which diagnostics would you capture? I am now taking jcmd Thread.print dumps while the JVM is stuck, to see who holds the lock. submitted by /u/Just-Vegetable3622 [link] [comments]

Spark writeStream inefficiencies for N sinks

In Spark Structured Streaming when the stream writes to a output table (sink) you will typically use writeStream (other option is foreachBatch) writeStream has many benefits (fault tolerant...checkpoint per sink etc) but one issue is that it will process the source N times where N is the amount of sinks/ output tables you have. So lets say for a bronze -> silver app, where it reads bronze, transforms the data (eg. explode) and then writes to 4 silver tables. In theory this will be inefficient as the source read, deserialize, explode will happen 4 times (once per sink) This happens as it is visible in the logs, 4x logs of source read etc This is the theory... has anyone experienced this in practice? is there actually a 4x slowdown or inefficency? Would introducing an intermediate table help as this would remain 1-1 with only one sink thus no issue.... but introduce extra write cost + read + extra storage layer What is best practice here? (foreachBatch is not an option due to any possible data loss) submitted by /u/BigDaddyData187 [link] [comments]

What Databricks feature do you use every day but couldn’t explain under the hood?

Mine used to be Structured Streaming checkpoints. I’d used them for months before I really understood what gets stored in there, and why deleting or reusing one quietly breaks a stream. I’m a trainer and I spend most of my week explaining this stuff, so post yours. I’ll try to answer every one: how it actually works underneath, no hand-waving. Bonus points if you say what specifically feels fuzzy: what it does, how it does it, or when to pick it over something else. submitted by /u/UnluckySugar610 [link] [comments]

Streaming On-Demand State Repartitioning is now GA

We are excited to announce that Streaming On-Demand State Repartitioning is now Generally Available for Structured Streaming and Spark Declarative Pipelines. This enables you to seamlessly change the number of partitions for your stateful streaming queries without losing the existing checkpoint. 📚Learn More Doc : https://docs.databricks.com/aws/en/structured-streaming/state-repartitioning Blog : https://www.databricks.com/blog/announcing-demand-state-repartitioning-apache-sparktm-structured-streaming-databricks Give it a try! submitted by /u/Ok-Brick-001 [link] [comments]

Read this if you use Streaming Tables in Lakeflow Spark Declarative Pipelines

🚀 We’re excited to announce that Lakeflow Spark Declarative Pipelines (SDP) now supports creating “vanilla” (i.e., non STREAMING) MANAGED TABLES and writing to them via one or more append flows , using the new CREATE TABLE ... FLOW ( SQL ) and create_table() (Python) APIs . What is this Beta? This Beta allows creating a managed table that is populated by append flows: CREATE TABLE ... FLOW (SQL) / create_table() + @append_flow (Python) create a managed table written by one or more flows. Fan multiple sources into one table — declare several flows targeting the same managed table. Full table surface works: partitioning, liquid clustering, expectations, row filters, table properties, and private (pipeline-local) tables. import_checkpoint on append_flow , which migrates an existing Structured Streaming workload into a pipeline without reprocessing the source — the flow imports the query's existing checkpoint and resumes from the last committed offset with state intact. Example (Python): from pyspark import pipelines as dp dp.create_table("combined") dp.append_flow(target="combined") def from_a(): return spark.readStream.table("source_a") u/dp.append_flow(target="combined") def from_b(): return spark.readStream.table("source_b") Example (SQL): CREATE TABLE events PARTITIONED BY (bucket) FLOW INSERT BY NAME SELECT id, bucket FROM STREAM read_files('abfss://my_path', format => 'json'); Where do we need help? We are in Beta, so there might be some rough edges. Please take this for a spin and share your feedback here . What’s next? Managed Tables support for other flow types (AutoCDC, Replace Using, and Replace Where) is coming soon! Learn more CREATE TABLE ... FLOW (SQL reference) — https://docs.databricks.com/aws/en/ldp/developer/ldp-sql-ref-create-table-flow create_table (Python reference) — https://docs.databricks.com/aws/en/ldp/developer/ldp-python-ref-create-table import_checkpoint on append_flow — https://docs.databricks.com/aws/en/ldp/developer/ldp-python-ref-append-flow Questions, feedback, or help: comment below or share feedback in the form: https://forms.gle/7bGP5FYN7P1Z4WP27 submitted by /u/SlightImagination250 [link] [comments]

Do small companies actually use Databricks?

Sometimes I feel like Databricks is way too expensive. It feels like using a huge truck to move a single grain of sand. My company needs real-time data, but our data volume simply does not justify the use of Spark Structured Streaming. Despite this, they are insisting we move to Databricks. I'm worried our data infrastructure costs will jump from $1,000/month to $5,000/month or more due to the running costs of Databricks SQL Warehouses. Currently, I use Azure Container Apps with KEDA and Python, which helps me manage scaling and keep costs low. We ingest into Event Hubs, use ADX (Azure Data Explorer) as our OLAP warehouse, and archive cold data in a data lake. With this setup, I manage to process all our data with very low latency. When I tested this on Databricks Structured Streaming, I actually got higher latency and much higher costs. Would love to know what you guys think. submitted by /u/Puzzled-Mail-9092 [link] [comments]

Automatic change data feed is now generally available!

With automatic CDF, Databricks computes row-level changes at read time using row tracking, rather than materializing those changes during every write. Use change data feed on Databricks | Databricks on AWS Why does that mattre? - Better write performance for MERGE INTO and UPDATE workloads - No need to enable CDF individually on every eligible table - Lower storage overhead compared with legacy CDF - The same familiar APIs still work: table_changes() and readChangeFeed - Works with batch processing, Structured Streaming, and Databricks-to-Databricks Delta Sharing For Delta Lake, the main requirements include: • Databricks Runtime 19 LTS+ • A managed table or external table in Delta Lake format with row tracking enabled And if you’re already using legacy CDF, migration is really simple.Once the table meets the requirements, disable legacy CDF: https://preview.redd.it/0dddizd442oh1.png?width=710&format=png&auto=webp&s=fddc536d6aeca423b2e418f4856ae06843393749 submitted by /u/szymon_dybczak [link] [comments]

What is a Checkpoint in Structured Streaming?

What are you guys using for data ingestion in Databricks?

I've mostly been using Auto Loader for file-based ingestion in Databricks, especially when there are continuously arriving files. It's been working pretty well so far, but I'm curious what others are using in their projects. For example, are you mainly using: 1.Auto Loader 2.Copy INTO 3.Structured Streaming 4.Batch jobs 5.Some external ingestion tool One thing I'm trying to understand better us where the trade-offs are. For a large number of files, does Auto Loader still make the most sense, or are these cases where something like COPY INTO is simplet and more cost-effective? Also, how are you handling things like schema evolution, duplicate files, failed records, and reprocessing? I'm mainly interested in what people are actually using in production. If you've tried multiple approaches, which one ended up being the best balanceof performance, reliability and cost for you? submitted by /u/Delulu62134 [link] [comments]

why micro-batching matters so much in Databricks Auto Loader and Structured Streaming

Building a Spark Streaming Real-Time Mode (RTM) Pipeline — Millisecond Streaming with Kafka

I recently built a fully working real-time transaction enrichment pipeline using PySpark RTM paired with Kafka, achieving end-to-end latency in the milliseconds. The article covers: \- Real-Time Mode (RTM) fundamentals \- Kafka integration with Spark Structured Streaming \- Millisecond-latency pipeline architecture \- Real-time transaction enrichment patterns Blog: https://blog.devgenius.io/building-a-spark-streaming-real-time-mode-rtm-pipeline-millisecond-streaming-with-kafka-dda74e9ef284