Guidance on Real-Time Streaming with Reconciliation and Domain Cutover Handling – Kafka → Databricks → Azure SQL Hyperscale

Question

Guidance on Real-Time Streaming with Reconciliation and Domain Cutover Handling – Kafka → Databricks → Azure SQL Hyperscale

Janice Chi 320

We are currently designing a real-time ingestion and reconciliation architecture using Databricks Structured Streaming that reads IBM DB2 CDC messages via Kafka and writes them to Azure SQL Hyperscale. The context is a healthcare data migration initiative involving four interdependent domains.

✅ Architecture Background:

Source: IBM DB2 (on-prem) with InfoSphere CDC pushing data into Kafka.

Ingestion Tool: Databricks Structured Streaming (Scala/PySpark).

Sink: Azure SQL Hyperscale (Main Table).

Cutover Behavior:

Catch-up completes per domain (e.g., Domain A).

  Domain A becomes live in Hyperscale only, while B, C, and D are active in both source and target during streaming.

We would appreciate your recommendations on the following technical design challenges:

Trigger Design for Structured Streaming What is the recommended practice for defining micro-batch triggers in Databricks Structured Streaming under production workloads?

Should the trigger interval be time-based (e.g., 1 minute, 2 minutes) or based on a record count threshold (if supported)?

  Which approach offers the most balanced trade-off between **latency**, **throughput**, and **cluster stability**, especially in healthcare-grade workloads with JSON messages?
  
  **Performing Reconciliation Before Write to Hyperscale** For each micro-batch, we must perform reconciliation (e.g., row counts or hash checks) before writing to Azure SQL Hyperscale.
  
     What methods are recommended to ensure this **reconciliation happens within the same batch** without introducing latency?
     
        Is it appropriate to implement this using `foreachBatch`, and if so, what considerations should be followed to avoid blocking ingestion?
        
        **Handling CDC Data Overlap During Domain Cutover** After domain cutover (e.g., Domain A), old records may still arrive in Kafka due to lag or late CDC entries.
        
           How should we design our streaming pipeline to **prevent re-inserting previously merged records** from the catch-up load?
           
              Would it be appropriate to use techniques such as **load timestamp filtering**, **watermarking**, or **deduplication by primary key + timestamp**?
              
                 Is there a need to pause or restart the streaming job during cutover, or can it handle this logic dynamically?
                 
                 **Data Write into Hyperscale with Minimal Latency** We need to ensure that, after reconciliation passes, each validated batch is written to Hyperscale with minimal delay.
                 
                    What is the best practice for writing from Databricks to Azure SQL Hyperscale in streaming mode?
                    
                       Should we consider a two-step approach (e.g., insert into staging + merge to main), or directly insert into the main table if transformations and merges are already applied upstream?

Venkat Reddy Navari 3,630 Reputation points Microsoft External Staff Moderator

2025-06-30T06:26:15.0166667+00:00

@Janice Chi We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.

Accepted answer

0 additional answers

Your answer

Venkat Reddy Navari 3,630 Reputation points Microsoft External Staff Moderator

2025-06-30T06:26:15.0166667+00:00

@Janice Chi We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.

Answer 1

Venkat Reddy Navari 3,630 Microsoft External Staff Moderator

Hi @Janice Chi Trigger Design in Databricks Structured Streaming: In most production setups, I’ve had the best results using a time-based trigger, like trigger(processingTime="1 minute"). It gives a good trade-off between low latency and stability—especially when you're dealing with JSON payloads from CDC like you are.

Record-count-based triggers aren’t really supported natively in Structured Streaming, and they tend to be harder to control at scale anyway. A one-minute interval gives your cluster time to process batches predictably, and it plays nicely with autoscaling if you’re using it.

If you notice lag or batch processing getting spiky, you can always fine-tune this, but a minute is a solid place to start.

Reconciliation Before Writing to Azure SQL: Yes, foreachBatch is the right place to do in-batch reconciliation. Inside that function, you can validate counts, do hash checks, or even compare expected vs. actual rows before pushing to Azure SQL.

Keep that logic efficient. Try to avoid joins or lookups that could slow the batch down. If you log audit results (e.g., counts, checksums) into a side table or log store, it’ll help you track issues without blocking the stream.

Just make sure whatever reconciliation you do stays within the same micro-batch this way you don’t break consistency when writing to Hyperscale.

Handling Late CDC Records After Domain Cutover: This is a common headache especially when you cut over Domain A, but Kafka still spits out some lagged messages.

Best options here:

Deduplicate by primary key + timestamp before writing.
Watermarking can help drop messages that are too late, though you need to balance latency tolerance.
Add cutover logic in your stream you can reference a "domain state" table (e.g., Domain A = live) and drop/ignore any old records that shouldn’t be processed anymore.

You don’t need to pause the stream during cutover. It’s better to handle this through logic build your stream to dynamically check cutover state and filter accordingly.

Writing to Azure SQL Hyperscale with Minimal Latency: In streaming scenarios, writing directly into your main tables can be risky especially if reconciliation or merge logic is still needed. A safer and more scalable pattern is:

Write to a staging table.
Run a stored proc or merge operation into your target tables.

This way, you can retry failed batches, audit changes, or even roll back bad loads more cleanly.

If you do choose to write directly into the main tables, just make sure you’ve already done all your business rules, transformations, and deduplication upstream. Also tune your JDBC settings (batch size, retries, etc.) so you’re not bottlenecked at write time.

I hope this information helps. Please do let us know if you have any further queries.

Kindly consider upvoting the comment if the information provided is helpful. This can assist other community members in resolving similar issues.

Janice Chi 320 Reputation points

2025-06-27T14:26:08.0166667+00:00

can u explian that if recon fails then how we are making sure that next microbatch is not "ON HOLD" - please explian with example and code -Keep that logic efficient. Try to avoid joins or lookups that could slow the batch down. If you log audit results (e.g., counts, checksums) into a side table or log store, it’ll help you track issues without blocking the stream.

Just make sure whatever reconciliation you do stays within the same micro-batch this way you don’t break consistency when writing to Hyperscale.

Venkat Reddy Navari 3,630 Microsoft External Staff Moderator

@Janice Chi By default, Databricks will continue processing the next micro-batch as long as you handle errors properly inside foreachBatch. If an exception is unhandled, it will block and retry the current batch.

Recommended Pattern

Use try/except in foreachBatch to catch reconciliation failures, log them, and skip writing allowing the stream to proceed:


def process_batch(df, batch_id):
    try:
        source_count = df.count()
        expected_count = 1000  # Replace with lookup from control table

        if source_count != expected_count:
            raise ValueError(f"Mismatch: Source={source_count}, Expected={expected_count}")

        # Proceed to write
        df.write.format("jdbc").option("url", jdbc_url).option("dbtable", "staging.table").mode("append").save()

    except Exception as e:
        # Log the error without halting the stream
        error_df = spark.createDataFrame([(batch_id, str(e))], ["batch_id", "error"])
        error_df.write.format("jdbc").option("url", jdbc_url).option("dbtable", "logging.recon_errors").mode("append").save()

        print(f"[Warning] Reconciliation failed for batch {batch_id}: {e}")

Attach this function to your stream:


df_streaming.writeStream.foreachBatch(process_batch).start()

Want to Stop Stream on Reconciliation Failures?

Let the exception propagate (remove the try/except). This will block or stop the stream until you resolve the issue.

Janice Chi 320 Reputation points

2025-07-01T12:07:34.56+00:00
We are using the foreachBatch construct to perform reconciliation (row count/hash checks) before writing each batch.

We have the following questions related to trigger behavior, batching, and processing latency:

Trigger Behavior – Time vs. Volume Based:

Given that we are receiving ~3000 CDC events per second in Kafka:

If we use .trigger(processingTime="1 minute"), will the stream trigger a micro-batch as soon as 1 minute passes, regardless of how many records arrived?

Can we define a condition like “trigger when either 1 minute OR 5000 records have arrived” (i.e., whichever comes first)? If not supported natively, are there recommended patterns to simulate such behavior?

ForeachBatch and Latency Implications:

Assume that:

In one micro-batch, we receive 180,000 events (from 1 min @ 3000/sec),

But in the next batch (e.g., load spike), we get 1.8 million records (30K/sec),

Then for 10 consecutive batches, this high volume continues.

Questions:

Since foreachBatch is sequential, does this mean that the next micro-batch will wait until the previous foreachBatch logic (including reconciliation and JDBC write) completes?

If so, will this cause ingestion latency, and how can this be mitigated

Defining "Near Real-Time" in Practical Terms:

In a healthcare-grade streaming system like ours:

What is considered an acceptable end-to-end ingestion latency from Kafka to Azure SQL?

Is it feasible to expect ingestion and write completion within 5 seconds per event, or should we plan for ~30s–60s delay per domain in practice?We are using the foreachBatch construct to perform reconciliation (row count/hash checks) before writing each batch.

Multi-Stream Design – One Near Real-Time Pipeline per Domain (B, C, D):

We are processing 4 domains in total. Domain A has completed cutover. For Domains B, C, and D, is it a best practice to set up separate Structured Streaming pipelines per domain, rather than handling all domains in a single stream?

If yes, what are the required changes at:

Databricks notebook/pipeline level (e.g., per-domain filters, checkpoints)?

**Kafka subscription level** (e.g., separate topics or consumer groups)? **Infrastructure scaling level** (e.g., per-domain cluster tuning)? Would this architecture help with better error isolation, scaling, and domain-specific cutover logic?
Venkat Reddy Navari 3,630 Reputation points Microsoft External Staff Moderator

2025-07-01T16:48:41.87+00:00
@Janice Chi

Trigger Behavior – Time vs. Volume:

.trigger(processingTime="1 minute") triggers once per minute, regardless of record volume.

Structured Streaming does not support “whichever comes first” triggers natively.

Workaround: Simulate volume triggers by:

Throttling at Kafka-level (e.g., maxOffsetsPerTrigger),

Capping inside foreachBatch (df.limit(...)) with caution to avoid drops.

For healthcare-grade consistency, time-based triggers (e.g., 1–2 min) are still preferred.

ForeachBatch & Ingestion Lag:

Yes, foreachBatch is sequential; next batch waits for the current one to finish (including JDBC write).

To minimize latency:

Optimize logic: avoid large joins or lookups.

Log reconciliation results asynchronously (e.g., to Delta/log table).

Consider a 2-step pipeline: stream to Delta, then use a separate job for validation + write to Hyperscale.

Realistic Latency Expectations:

Expect 30–60 sec E2E latency per domain with reconciliation and JDBC writes.

Sub-10s latency is rare unless you skip validation or use Delta + async downstream processing.

Multi-Stream (Per-Domain) Design:

Recommended: one stream per domain (B, C, D).

Better isolation, fault tolerance, tuning, and domain-specific cutover logic.

Adjustments:

Per-domain notebook filters & checkpoints.

Kafka: either separate topics or shared topic + consumer group filtering.

Cluster scaling: size per domain load profile.

Reconciliation Failures – Keep Stream Moving:

Use try/except inside foreachBatch to log errors but avoid blocking the stream:

def process_batch(df, batch_id): try: if not reconcile(df): raise Exception("Recon failed") df.write.format("jdbc").option(...).save() except Exception as e: log_error(batch_id, e) # Write to Delta or JDBC log table

This keeps the stream healthy even if a batch fails recon.

Hope this helps. If this answers your query, do click Accept Answer and Yes for was this answer helpful. And, if you have any further query do let us know.

Share via

Guidance on Real-Time Streaming with Reconciliation and Domain Cutover Handling – Kafka → Databricks → Azure SQL Hyperscale

0 additional answers

Your answer