Big Data Resources

DZone's Featured Big Data Resources

Top 10 Best Places to Prepare for Your Next Data Engineer Interview

By Rahul Han

Landing a data engineering role means clearing a gauntlet that no other software discipline has to face all at once: airtight SQL, production-grade Python, data modeling instincts, distributed-compute fluency (Spark, warehouses, ETL), and system design that has to survive real data volume. Generic coding prep barely scratches the surface, and "just grind LeetCode" advice falls apart the moment an interviewer asks you to model a slowly changing dimension or reason about a skewed join. So we did the work. We evaluated the resources data engineers actually use, judged on five things that matter: relevance to the DE interview loop, depth of practice, realism of the questions, feedback quality, and price. Below is the ranked list. A quick note on methodology: this ranking favors resources that target the data engineering loop specifically, not generic algorithm grinding. That bias is intentional, and it is why the order may surprise you. 1. DataDriven.io Most "interview prep" platforms were built for generic SWE roles and bolt on a SQL section as an afterthought. This one was built from the ground up for the data engineering loop. The catchphrase you will hear repeated in DE communities is that DataDriven.io is LeetCode for data engineers, and it fits: instead of inverting binary trees, you are writing window functions against realistic schemas, designing star schemas, debugging an ETL transform, and reasoning about partitioning, all in an in-browser SQL and Python sandbox that runs your query against real data and tells you exactly where it broke. It is also the rare place where the whole product is built for the job rather than adjacent to it, which is why datadriven.io is great for data engineer interview prep specifically: SQL practice that ramps to multi-CTE analytics, a deep set of Python practice problems, plus data modeling, dimensional modeling, PySpark, and system-design tracks, with execution-based feedback and a difficulty curve that reaches the staff-level questions that actually separate offers from rejections. Verdict: The most targeted, realistic data engineering interview practice available today. Earns the top spot. 2. "Cracking the Coding Interview" (the book, by Gayle Laakmann McDowell) A deserved classic, and intentionally a book rather than a website. CTCI is still the best single artifact for understanding how technical interviews are actually structured: how the conversation flows, how to think out loud so the interviewer can follow your reasoning, how to recover when you get stuck, and how to handle the behavioral and negotiation segments that strong candidates routinely fumble. Most people lose offers not because they could not solve the problem but because they could not show their work, and this book is the canonical fix for that. Where it falls short for our purposes is scope. It will not teach you windowed SQL, slowly changing dimensions, or how to design a lakehouse, and its algorithm focus skews toward generalist software roles rather than the data engineering loop. The data structures and big-O chapters are still worth a pass because algorithm screens do show up, but treat them as a refresher, not your main event. Read CTCI once early in your prep to fix your interview mechanics, internalize the communication patterns, then spend the rest of your time on hands-on, domain-specific platforms. Verdict: Essential reading for interview mechanics; not a substitute for domain practice. 3. "Designing Data-Intensive Applications" (the book, by Martin Kleppmann) If CTCI teaches you how to interview, "DDIA" teaches you what a data engineer is actually supposed to know. Replication, partitioning, consistency models, batch versus stream processing, storage engine internals, the failure modes of distributed systems: this is the conceptual backbone of nearly every data engineering system design round. When an interviewer asks why you would choose a log-structured merge tree over a B-tree, or how you would keep two datastores in sync without losing events, the answers live in these pages. It is dense, and it is emphatically not an interview drill book. You will not find practice questions, and you cannot cram it the night before. What it gives you instead is judgment: the candidate who has internalized DDIA answers "how would you design this pipeline" with the calm of someone who has already thought through the tradeoffs, names the failure cases before being prompted, and explains why a choice holds up under real data volume. Read it slowly over weeks, ideally early in your prep, and pair it with a hands-on platform so the concepts attach to actual queries and schemas rather than floating as theory. Verdict: The definitive conceptual reference. Read it slowly, alongside real practice. 4. LeetCode The default destination, and it earns its spot for one practical reason: the Database problem set is sizable, the algorithm catalog is enormous, and the platform's brand means a large share of companies still pull their initial coding screen straight from it. If your target company is known to run a generic algorithm round before the data-specific rounds, you need exposure here, and the sheer volume of problems plus community discussion means you will rarely be surprised by a pattern you have never seen. The catch for data engineers is that LeetCode was built for the algorithm interview, not the DE loop. Its SQL section is genuinely solid but secondary; the questions are puzzle-shaped rather than drawn from real schemas, and you will not find data modeling, ETL design, dimensional modeling, or Spark anywhere on the platform. There is also a real failure mode here: candidates over-invest in LeetCode because it is comfortable and gamified, then walk into a DE loop under-practiced on the things that actually decide it. Use it deliberately to clear the algorithm gate and to keep your raw coding sharp, then move the bulk of your hours to resources that target data engineering directly. Verdict: Necessary for the algorithm screen; thin for the data-engineering-specific rounds. 5. HackerRank HackerRank is where a surprising number of companies host their take-home and timed online assessments, so practicing in its environment carries a payoff most resources cannot offer: you get comfortable with the exact editor, the exact test-case runner, and the exact time-pressure UI you may actually be scored in. For an assessment you cannot retake, that familiarity is worth real points, because fighting an unfamiliar interface while the clock runs is a self-inflicted way to lose. Its SQL and problem-solving tracks are beginner-friendly, well-structured, and free to work through. The ceiling, though, is lower than you want for a senior DE loop. The problems lean academic and self-contained rather than job-realistic, the SQL rarely reaches the messy multi-table analytics that real interviews probe, and there is nothing on modeling, pipelines, or system design. The smart way to use HackerRank is as format rehearsal: run a few timed sets so the assessment environment feels routine, then build your actual depth somewhere that mirrors the work. Do not let a green checkmark on an easy problem set convince you that you are loop-ready. Verdict: Great for getting comfortable with the testing environment; limited depth. 6. SQLZoo A long-running, completely free interactive SQL tutorial that runs entirely in the browser with no signup, no setup, and no paywall. It walks you from SELECT basics through joins, grouping, subqueries, and window functions, with short hands-on exercises after each concept so you are writing real queries from the first lesson rather than just reading about them. For anyone whose SQL has gone rusty, or who learned it informally and has gaps they cannot quite name, it is the most painless way to rebuild muscle memory before stepping up to interview-grade problems. It is a teaching tool, not an interview platform, and you should treat it as exactly that. The problems stay introductory, the datasets are small and tidy, and there is nothing on data modeling, ETL, pipelines, or system design — the parts of the loop that actually separate data engineers from analysts. Its value is as a fast diagnostic and warm-up: work through the sections that feel shaky, confirm your fundamentals are solid, then graduate to harder, execution-based practice against realistic schemas. Linger here too long, and you will plateau well below where a real interview will push you. Verdict: A friendly free SQL primer; foundational rather than interview-level. 7. "Python for Data Analysis" (by Wes McKinney) Written by the creator of pandas, this is the reference for the kind of data-wrangling Python that shows up constantly in DE take-homes and pairing rounds: reshaping, grouping and aggregating, merging on imperfect keys, handling missing values, parsing dates, and cleaning the kind of messy tabular data that never looks like a tidy LeetCode input. Many data engineering interviews quietly assume this fluency, then hand you a notebook and a dirty CSV and watch how you move; if your Python is sharp on algorithms but clumsy on real data manipulation, this book is exactly the gap-closer. It is a library-and-technique book, not interview prep, and it will not touch SQL, data modeling, distributed compute, or system design. There are also no interview questions to grind, which is fine, because its job is to make the tools second nature so that during a timed exercise you are reasoning about the problem instead of fumbling for the right pandas idiom. Read the chapters on data loading, cleaning, and group operations, keep it nearby as a reference, then go apply the techniques in hands-on practice against problems that actually resemble the job. Verdict: The definitive practical Python reference for data work; not a drill book. 8. "Fundamentals of Data Engineering" (the book, by Joe Reis & Matt Housley) Another deliberate book pick, and the best single survey of the modern data engineering lifecycle: generation, ingestion, storage, transformation, and serving, plus the cross-cutting concerns like orchestration, data quality, and governance that interviewers increasingly probe. Where DDIA goes deep on systems internals, this book goes broad on how the pieces fit together into a working data platform, which is precisely the framing you want for the "walk me through how you'd build X" and "what would you consider before choosing this approach" portions of a loop. It is a framework-and-vocabulary book, not a practice book, and that is both its strength and its limit. It will give you the mental model and the shared language to discuss tradeoffs like a practitioner, which makes you sound, accurately, like someone who understands the field. But it contains no exercises, so reading it alone will not build the hands-on skill an interviewer also tests. Use it to organize everything you know into a coherent lifecycle, fill the conceptual gaps, then go write the queries and design the schemas somewhere that gives you real feedback. Verdict: The best lifecycle overview in print; conceptual, not hands-on. 9. Mode SQL Tutorial A free, well-regarded interactive SQL tutorial built by an analytics company, which shows in its framing: it teaches SQL the way analysts and engineers actually use it, oriented around answering real questions from data rather than solving abstract puzzles. It runs in the browser, takes you from the basics through intermediate analytics queries including aggregation and the early window-function territory, and the explanations are unusually clear about why a query is shaped the way it is. For someone shoring up SQL foundations before diving into harder problems, it is one of the cleanest no-cost on-ramps available. Like SQLZoo, it is a tutorial rather than an interview-prep platform, so it stops well short of the difficulty a real DE loop will throw at you, and it covers none of the modeling, pipeline, or system-design ground. It is best read as a companion to a hands-on platform: use Mode to internalize the analytical mindset and clean up your SQL fundamentals, then take that foundation into execution-based practice where the problems are harder, the schemas messier, and the feedback tells you exactly where your query went wrong. Verdict: A clean free SQL on-ramp; foundational rather than interview-level. 10. Pramp/Interviewing.io (mock interviews) Rounding out the list: peer and expert mock interviews. All the solo practice in the world cannot reproduce the specific pressure of explaining your reasoning out loud to a real human while a clock runs and someone is judging you, and that pressure is exactly where otherwise-prepared candidates fall apart. A handful of mock loops surface the weaknesses you cannot see in yourself: the long silences, the jumping to code before clarifying the question, the inability to narrate a tradeoff. Pramp pairs you with peers for free, while Interviewing.io connects you with experienced interviewers, often anonymously, for higher-fidelity feedback. The honest limitation is supply and specificity. Data-engineering-focused interviewers are scarcer than generalist software ones, so depending on availability, you may land in an algorithm or general system-design mock that only partially mirrors a true DE loop. That is still worth doing, because the communication skills, the structure, the clarifying questions, the calm narration, transfer directly regardless of the exact problem. Schedule one or two once your technical prep is underway, treat the feedback as data, and fix the delivery habits well before the interview that counts. Verdict: Best for rehearsing delivery and nerves; DE-specific matches can be hit-or-miss. How to Actually Use This List You do not need all ten. A focused plan beats a scattered one: Build the foundation. Skim CTCI for interview mechanics and start DDIA for concepts.Do the reps where it counts. Spend the bulk of your time on hands-on, DE-shaped practice that maps directly onto what you will be asked (see #1).Patch specific gaps. Use LeetCode for the algorithm screen, SQLZoo or the Mode tutorial to shore up SQL, and a mock interview or two to rehearse out loud. The candidates who get offers are not the ones who consumed the most content. They are the ones who practiced the actual job. Pick the resources that put you closest to it, start today, and write more queries than you read. Good luck with your loop. More

Building Production-Grade Delta Lake Pipelines With Apache Spark on Databricks

By Jubin Abhishek Soni

CORE

Why Delta Lake? Apache Parquet on cloud storage was a great first step for data lakes — but it left engineers dealing with a painful set of problems in production: No ACID transactions — concurrent reads/writes could corrupt data silentlySchema drift — nothing stopped upstream systems from changing column typesNo deletes or updates — GDPR compliance meant rewriting entire partitionsPainful failure recovery — half-written data after a job crash became your problem Delta Lake solves all of this by sitting on top of Parquet and adding a transaction log (_delta_log/) that records every operation atomically. On Databricks, Delta is the default table format, deeply integrated with Apache Spark, Auto Optimize, and the Photon execution engine. The Medallion Architecture The medallion (or multi-hop) architecture organizes data into three progressive refinement layers. Each layer has a clear contract with the layers around it. Each layer has a distinct responsibility: LayerAliasPurposeRetentionBronzeRawLand data as-is, preserve source fidelityYears (audit trail)SilverCleansedDeduplicate, validate, type-cast, conform schemasMonthsGoldAggregatedBusiness-level KPIs, domain-specific aggregatesMonths–Years The key design principle is: never skip a layer. Debugging a production incident is infinitely easier when you can replay from raw Bronze data. Delta Lake Internals: The Transaction Log Before writing a single line of code, it's worth understanding how Delta Lake achieves ACID semantics. Every write operation (INSERT, UPDATE, DELETE, MERGE) produces a new JSON entry in _delta_log/. This log-based approach gives you time travel for free — querying VERSION AS OF 2 simply replays only the log entries up to that point. Checkpoints every 10 commits keep read performance snappy even with thousands of versions. Setting Up Your Databricks Environment All code here runs on Databricks Runtime 13.x+ (which ships Delta Lake 2.x). For local dev, use the delta-spark package. # Databricks notebook — Runtime 13.3 LTS or higher # Delta Lake is pre-installed; no pip install needed from pyspark.sql import SparkSession from pyspark.sql import functions as F from pyspark.sql.types import ( StructType, StructField, StringType, LongType, DoubleType, TimestampType, BooleanType ) from delta.tables import DeltaTable # On Databricks, SparkSession is pre-created as `spark` # For local testing: # spark = (SparkSession.builder # .appName("medallion-pipeline") # .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") # .config("spark.sql.catalog.spark_catalog", "spark.sql.delta.catalog.DeltaCatalog") # .getOrCreate()) # Unity Catalog paths (recommended for production) CATALOG = "prod" BRONZE_DB = f"{CATALOG}.bronze" SILVER_DB = f"{CATALOG}.silver" GOLD_DB = f"{CATALOG}.gold" # DBFS / external storage path (for raw file landing) RAW_LANDING = "abfss://[email protected]/events/" Building the Bronze Layer Bronze is your append-only raw ingestion layer. The goal is landing data with zero transformation — preserve everything, even malformed records. Schema is inferred or declared loosely. # ── Bronze Ingestion ────────────────────────────────────────────────────────── # Using Auto Loader (cloudFiles) — the Databricks-native incremental ingest # mechanism. It tracks which files have been processed via a checkpoint, # so re-running never double-injects data. raw_event_schema = StructType([ StructField("event_id", StringType(), True), StructField("user_id", StringType(), True), StructField("event_type", StringType(), True), StructField("event_ts", StringType(), True), # keep as string in bronze StructField("properties", StringType(), True), # raw JSON blob StructField("session_id", StringType(), True), StructField("platform", StringType(), True), ]) bronze_stream = ( spark.readStream .format("cloudFiles") .option("cloudFiles.format", "json") .option("cloudFiles.schemaLocation", "/checkpoints/bronze_events_schema") .option("cloudFiles.inferColumnTypes", "false") # keep raw types .schema(raw_event_schema) .load(RAW_LANDING) # Enrich with ingestion metadata — critical for debugging .withColumn("_ingest_ts", F.current_timestamp()) .withColumn("_source_file", F.input_file_name()) .withColumn("_ingest_date", F.to_date(F.current_timestamp())) ) ( bronze_stream.writeStream .format("delta") .outputMode("append") .option("checkpointLocation", "/checkpoints/bronze_events") .option("mergeSchema", "true") # allow new columns from upstream .partitionBy("_ingest_date") # partition for incremental silver reads .trigger(availableNow=True) # run-once trigger (for scheduled jobs) .tableCheckpoint(f"{BRONZE_DB}.events_raw") .toTable(f"{BRONZE_DB}.events_raw") ) Pro tip: Always add _ingest_ts and _source_file metadata columns in Bronze. When an upstream system sends corrupt data at 3 AM, these columns tell you exactly which file batch caused it. Transforming to Silver Silver is where the real engineering work happens: deduplication, type casting, schema validation, and applying business rules. We also handle CDC (Change Data Capture) upserts here using Delta's MERGE. # ── Silver Transformation ───────────────────────────────────────────────────── def transform_bronze_to_silver(bronze_df): """ Apply cleansing and conforming rules to raw Bronze events. Returns a Silver-ready DataFrame with enforced schema. """ return ( bronze_df # 1. Parse timestamps properly .withColumn("event_ts", F.to_timestamp("event_ts", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")) # 2. Parse the raw JSON properties blob into a struct .withColumn("props", F.from_json( F.col("properties"), schema="page_url STRING, referrer STRING, duration_ms LONG, revenue DOUBLE" )) .drop("properties") # 3. Normalize platform values .withColumn("platform", F.lower(F.trim(F.col("platform")))) .withColumn("platform", F.when( F.col("platform").isin("ios", "android"), F.col("platform") ).when( F.col("platform").isin("web", "browser", "desktop"), F.lit("web") ).otherwise(F.lit("unknown"))) # 4. Filter out test/internal traffic .filter(~F.col("user_id").startswith("test_")) .filter(F.col("event_id").isNotNull()) # 5. Deduplicate within the micro-batch (window by event_id) .dropDuplicates(["event_id"]) # 6. Add Silver metadata .withColumn("_silver_ts", F.current_timestamp()) .withColumn("event_date", F.to_date("event_ts")) # partition key ) # Upsert into Silver using MERGE (handles late-arriving / duplicate events) def upsert_to_silver(micro_batch_df, batch_id): micro_batch_df = transform_bronze_to_silver(micro_batch_df) silver_table = DeltaTable.forName(spark, f"{SILVER_DB}.events_clean") ( silver_table.alias("target") .merge( micro_batch_df.alias("source"), "target.event_id = source.event_id" # dedup key ) .whenMatchedUpdateAll() # update if record arrived late with corrections .whenNotMatchedInsertAll() # insert new records .execute() ) # Stream from Bronze → Silver using foreachBatch ( spark.readStream .format("delta") .option("readChangeFeed", "true") # CDF — only process new Bronze rows .table(f"{BRONZE_DB}.events_raw") .writeStream .foreachBatch(upsert_to_silver) .option("checkpointLocation", "/checkpoints/silver_events") .trigger(availableNow=True) .start() ) Aggregating to Gold Gold tables are business-ready aggregates consumed directly by BI tools, dashboards, and ML feature pipelines. They are typically batch-refreshed on a schedule. # ── Gold Aggregation ────────────────────────────────────────────────────────── daily_revenue = ( spark.table(f"{SILVER_DB}.events_clean") .filter(F.col("event_type") == "purchase") .filter(F.col("event_date") >= F.date_sub(F.current_date(), 90)) # rolling 90d .groupBy("event_date", "platform") .agg( F.sum("props.revenue").alias("total_revenue"), F.countDistinct("user_id").alias("unique_buyers"), F.count("event_id").alias("transaction_count"), F.avg("props.duration_ms").alias("avg_session_duration_ms"), ) .withColumn("revenue_per_buyer", F.round(F.col("total_revenue") / F.col("unique_buyers"), 2)) .withColumn("_gold_ts", F.current_timestamp()) ) # Overwrite with replaceWhere — only touch the last 90 days, not the full table ( daily_revenue.write .format("delta") .mode("overwrite") .option("replaceWhere", "event_date >= date_sub(current_date(), 90)") .saveAsTable(f"{GOLD_DB}.daily_revenue") ) Z-Ordering and Data Skipping Z-ordering is Databricks' multi-dimensional clustering technique. It co-locates related data within the same set of Parquet files, so Spark can skip irrelevant files entirely during queries — without the overhead of strict partitioning. -- Run OPTIMIZE + ZORDER after significant writes -- This rewrites data files to cluster on the most commonly filtered columns OPTIMIZE prod.silver.events_clean ZORDER BY (user_id, event_date, event_type); -- Check how many files were skipped in your last query -- (run immediately after a SELECT with filters) SELECT operation, operationMetrics['numFilesAdded'] AS files_added, operationMetrics['numFilesRemoved'] AS files_removed, operationMetrics['numRemovedBytes'] AS bytes_removed FROM ( DESCRIBE HISTORY prod.silver.events_clean ) WHERE operation = 'OPTIMIZE' ORDER BY timestamp DESC LIMIT 5; Rule of thumb: Z-order on your top 3–4 most-filtered columns. Beyond that, the clustering benefit diminishes and OPTIMIZE runtimes grow significantly. Never Z-order on partition columns — they're already physically separated. Optimized Spark Writes Poorly tuned Spark writes are the #1 cause of small-file problems in Delta Lake. Here's a production-hardened write configuration: # ── Write Configuration Reference ──────────────────────────────────────────── SILVER_WRITE_CONFIG = { # Coalesce output files to ~128MB each (avoids small-file explosion) "spark.sql.shuffle.partitions": "200", # tune to cluster size "spark.databricks.delta.optimizeWrite.enabled": "true", # auto bin-packing "spark.databricks.delta.autoCompact.enabled": "true", # background compaction # Target file size for Auto Optimize "spark.databricks.delta.optimizeWrite.binSize": "134217728", # 128 MB in bytes # Enable deletion vectors (Databricks 12.2+) — soft-deletes without file rewrites "spark.databricks.delta.enableDeletionVectors": "true", } # Apply at the session level for the pipeline job for k, v in SILVER_WRITE_CONFIG.items(): spark.conf.set(k, v) # For partitioned tables: control output file count per partition ( silver_df .repartition(F.col("event_date")) # one task group per date partition .write .format("delta") .mode("overwrite") .option("dataChange", "true") .option("overwriteSchema", "false") # never silently change schema in prod .partitionBy("event_date") .saveAsTable(f"{SILVER_DB}.events_clean") ) Pipeline Comparison Table Here's how different ingestion patterns stack up for common production scenarios on Databricks: PatternLatencyThroughputDedup SupportBest ForAuto Loader + AppendNear-real-timeVery High❌ NoEvent logs, immutable streamsAuto Loader + MERGENear-real-timeHigh✅ YesCDC, late-arriving eventsBatch COPY INTOMinutesHigh❌ NoScheduled file ingestionStructured Streaming + foreachBatchSecondsMedium✅ YesComplex stateful pipelinesDelta Live Tables (DLT)ConfigurableHigh✅ Yes (expectations)Declarative, managed pipelinesMERGE only (batch)MinutesLow–Medium✅ YesSmall-to-medium upsert volumes DLT note: Delta Live Tables is Databricks' managed pipeline framework that handles the orchestration, monitoring, and retry logic described above declaratively. For teams starting fresh, DLT is worth evaluating before building the plumbing manually. Key Takeaways Medallion architecture separates concerns cleanly: Bronze for fidelity, Silver for correctness, Gold for consumption.Delta's transaction log is the foundation of all ACID guarantees — understanding it helps you debug merge conflicts, time travel, and VACUUM safely.Auto Loader is the right default for cloud file ingestion on Databricks — it handles exactly-once semantics and schema evolution automatically.MERGE with foreachBatch is the idiomatic pattern for deduplication and CDC in Spark Structured Streaming.Z-ORDER + Auto Optimize should be standard practice for Silver and Gold tables that receive frequent queries with selective filters.Deletion Vectors (Databricks 12.2+) make point deletes significantly cheaper — enable them for tables with GDPR or compliance requirements. References Delta Lake Documentation — Delta Lake Transaction Log — The official deep dive into how the _delta_log works internally.Databricks — Medallion ArchitectureDatabricks — Auto Loader (cloudFiles)Databricks — Delta Lake OPTIMIZE and Z-OrderingDatabricks — Auto Optimize (Optimized Writes + Auto Compaction)Databricks — Deletion VectorsDatabricks — Delta Live Tables OverviewStructured Streaming + foreachBatch — Apache Spark Docs"The Delta Lake Paper" — VLDB 2020 (Armbrust et al.)Databricks Blog — Diving Into Delta Lake: Unpacking the Transaction Log More

Azure Databricks for Scalable MLOps and Feature Engineering With Apache Spark, Delta Lake, and MLflow

By Jubin Abhishek Soni

CORE

Real-Time AI Feature Engineering With Spark Structured Streaming and Databricks Feature Store

By Jubin Abhishek Soni

CORE

From Polling to PubSub: Building an Asynchronous OPC UA Stack in Python

By Harshith Narasimhan Srivatsa

Dead Letter Queue Patterns in Apache Flink: Handling Poison Messages Without Stopping Your Stream

Streaming systems usually fail in one of two ways: Loudly, when infrastructure breaksQuietly, when one bad record keeps replaying until the pipeline is effectively dead The second failure mode is more dangerous because it often starts with something small: malformed JSON, an unexpected schema change, a missing required field, or a downstream timeout that was never handled correctly. In Apache Flink, one unhandled exception can trigger a restart. If the same poison message is still sitting in Kafka after recovery, the job reads it again, fails again, restarts again, and enters a loop. At that point, the pipeline is technically "recovering," but operationally it is down. This is exactly why production Flink jobs need a Dead Letter Queue (DLQ) strategy from day one. A proper DLQ pattern does three things: Isolates bad records so they do not stop good onesCaptures enough failure context to debug the issue laterPreserves replayability so quarantined records can be reprocessed after the root cause is fixed Anything less is not really a DLQ. It is either silent data loss or delayed outage. In this article, I will walk through the most practical DLQ patterns for Apache Flink 1.18: Side outputs as the core DLQ primitiveRetry with exponential backoff for transient failuresTiered DLQ routing by error classKafka and S3 sink patternsMetrics and alertingReplay with a dedicated reprocessing jobA PyFlink version of the side output pattern The goal is simple: a bad message should never silently disappear, and it should never silently stop the stream. Why Poison Messages Break Otherwise Healthy Pipelines A poison message is any record that consistently fails processing. Typical examples include: Malformed JSONIncompatible schema versionsMissing required fieldsInvalid business valuesRecords that trigger unexpected code pathsMessages that repeatedly fail downstream enrichment calls Without DLQ handling, the failure path usually looks like this: The record enters the pipelineDeserialization or validation throws an exceptionThe operator failsFlink restarts from the last checkpointThe same record is consumed againThe same exception happens again That loop can continue indefinitely. The result is predictable: Throughput drops to zeroDownstream consumers starveCheckpoint recovery does not helpOn-call engineers get paged for a problem caused by one record This is why DLQ handling is not just an error-handling convenience. It is a core reliability pattern. What a DLQ Should Look Like in Flink In a streaming architecture, a DLQ is a durable destination for records that could not be processed successfully. For Flink, that means the DLQ record should usually include: Raw payloadError typeError messageStack trace or summarized failure contextFailure timestampSource metadata such as topic, partition, or offset when available That information matters because a DLQ is only useful if someone can answer two questions later: Why did this record fail?How do I replay it safely once the issue is fixed? If you only log the exception, you lose replayability. If you only store the payload, you lose debugging context. If you drop the record entirely, you lose both. So the design target is not "catch exceptions." The design target is durable, observable, replayable failure handling. Pattern 1: Use Side Outputs as the Core DLQ Primitive The most natural DLQ mechanism in Flink is the side output. A side output allows one operator to emit records to multiple streams: The main stream for successful recordsOne or more side streams for failures, late data, or quarantined records That makes it the right primitive for DLQ routing. Define the DLQ Envelope and Output Tag Java import org.apache.flink.util.OutputTag; import org.apache.flink.streaming.api.functions.ProcessFunction; import org.apache.flink.util.Collector; public static final OutputTag<DeadLetterRecord> DLQ_TAG = new OutputTag<DeadLetterRecord>("dead-letter-queue") {}; public record DeadLetterRecord( String rawPayload, String errorType, String errorMessage, String stackTrace, long failedAtEpochMs, String sourceTopicPartition, long sourceOffset ) {} The important point here is that the DLQ record is not just the failed payload. It is an envelope that preserves enough context for triage and replay. Route Failures Inside a ProcessFunction Java public class EntityEventProcessor extends ProcessFunction<String, EntityEvent> { @Override public void processElement( String rawMessage, Context ctx, Collector<EntityEvent> out) { try { EntityEvent event = parseAndValidate(rawMessage); out.collect(event); } catch (JsonParseException e) { ctx.output(DLQ_TAG, new DeadLetterRecord( rawMessage, "JSON_PARSE_FAILURE", e.getMessage(), getStackTrace(e), System.currentTimeMillis(), ctx.element().toString(), -1L )); } catch (SchemaValidationException e) { ctx.output(DLQ_TAG, new DeadLetterRecord( rawMessage, "SCHEMA_VALIDATION_FAILURE", e.getMessage(), getStackTrace(e), System.currentTimeMillis(), ctx.element().toString(), -1L )); } catch (Exception e) { ctx.output(DLQ_TAG, new DeadLetterRecord( rawMessage, "UNKNOWN_FAILURE", e.getMessage(), getStackTrace(e), System.currentTimeMillis(), ctx.element().toString(), -1L )); } } private EntityEvent parseAndValidate(String raw) throws JsonParseException, SchemaValidationException { EntityEvent event = objectMapper.readValue(raw, EntityEvent.class); if (event.entityId() == null || event.entityId().isBlank()) { throw new SchemaValidationException("entityId is required"); } if (event.timestamp() <= 0) { throw new SchemaValidationException("timestamp must be positive"); } return event; } } This is the minimum viable DLQ pattern, and it already solves the most important operational problem: bad records no longer stop good ones. Wire the Main Stream and DLQ Stream Java StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<String> kafkaSource = env .fromSource(buildKafkaSource(), WatermarkStrategy.noWatermarks(), "entity-events-source"); SingleOutputStreamOperator<EntityEvent> processed = kafkaSource.process(new EntityEventProcessor()); DataStream<EntityEvent> goodEvents = processed; DataStream<DeadLetterRecord> deadLetters = processed.getSideOutput(DLQ_TAG); goodEvents.sinkTo(buildDownstreamKafkaSink()); deadLetters.sinkTo(buildDlqKafkaSink()); env.execute("Entity Resolution Pipeline"); If you do nothing else, do this. Side outputs should be the default DLQ foundation in Flink. Pattern 2: Retry Transient Failures Before Escalating to DLQ Not every failure belongs in the DLQ immediately. Some failures are transient: A downstream service is temporarily unavailableA database call times outAn external API is rate-limitedA network dependency is briefly unstable If you send all of those directly to the DLQ, you create noise and bury the truly bad records. The better pattern is: Retry transient failures a limited number of timesUse exponential backoffEscalate to DLQ only after retries are exhausted Retry With KeyedProcessFunction and Timers Java public class RetryingEnrichmentProcessor extends KeyedProcessFunction<String, EntityEvent, EnrichedEvent> { private static final int MAX_RETRIES = 3; private static final long BASE_BACKOFF_MS = 500L; private transient ValueState<Integer> retryCountState; private transient ValueState<EntityEvent> pendingEventState; @Override public void open(Configuration parameters) { retryCountState = getRuntimeContext().getState( new ValueStateDescriptor<>("retry-count", Integer.class)); pendingEventState = getRuntimeContext().getState( new ValueStateDescriptor<>("pending-event", EntityEvent.class)); } @Override public void processElement( EntityEvent event, Context ctx, Collector<EnrichedEvent> out) throws Exception { try { EnrichedEvent enriched = callEnrichmentService(event); retryCountState.clear(); pendingEventState.clear(); out.collect(enriched); } catch (TransientServiceException e) { int retries = retryCountState.value() == null ? 0 : retryCountState.value(); if (retries >= MAX_RETRIES) { retryCountState.clear(); pendingEventState.clear(); ctx.output(DLQ_TAG, new DeadLetterRecord( event.toString(), "MAX_RETRIES_EXCEEDED", "Failed after " + MAX_RETRIES + " retries: " + e.getMessage(), getStackTrace(e), System.currentTimeMillis(), ctx.getCurrentKey(), -1L )); } else { retryCountState.update(retries + 1); pendingEventState.update(event); long backoffMs = BASE_BACKOFF_MS * (long) Math.pow(2, retries); ctx.timerService().registerProcessingTimeTimer( System.currentTimeMillis() + backoffMs ); } } catch (PoisonMessageException e) { ctx.output(DLQ_TAG, new DeadLetterRecord( event.toString(), "POISON_MESSAGE", e.getMessage(), getStackTrace(e), System.currentTimeMillis(), ctx.getCurrentKey(), -1L )); } } @Override public void onTimer( long timestamp, OnTimerContext ctx, Collector<EnrichedEvent> out) throws Exception { EntityEvent pending = pendingEventState.value(); if (pending == null) return; try { EnrichedEvent enriched = callEnrichmentService(pending); retryCountState.clear(); pendingEventState.clear(); out.collect(enriched); } catch (TransientServiceException e) { int retries = retryCountState.value(); if (retries >= MAX_RETRIES) { retryCountState.clear(); pendingEventState.clear(); ctx.output(DLQ_TAG, new DeadLetterRecord( pending.toString(), "MAX_RETRIES_EXCEEDED", "Timer retry exhausted: " + e.getMessage(), getStackTrace(e), System.currentTimeMillis(), ctx.getCurrentKey(), -1L )); } else { retryCountState.update(retries + 1); long backoffMs = BASE_BACKOFF_MS * (long) Math.pow(2, retries); ctx.timerService().registerProcessingTimeTimer( timestamp + backoffMs ); } } } } Why This Works Especially Well in Flink This pattern is stronger in Flink than in many other stream processors because timers and state are checkpointed. That means: Retry counters survive restartsPending events survive restartsScheduled retries resume after recovery In other words, the retry workflow itself is fault-tolerant. That is exactly what you want when handling transient failures in a long-running stream. Pattern 3: Split the DLQ by Failure Type Once a pipeline matures, a single DLQ topic usually becomes too coarse. Schema failures, business validation failures, exhausted retries, and unknown exceptions all end up mixed together. That makes triage slower and replay harder. A better pattern is to classify failures and route them to separate DLQ streams. Define Failure Tiers Java public enum DlqTier { TRANSIENT_EXHAUSTED, SCHEMA_INVALID, BUSINESS_RULE, UNKNOWN } Route by Exception Class Java public class TieredDlqRouter extends ProcessFunction<String, EntityEvent> { @Override public void processElement( String raw, Context ctx, Collector<EntityEvent> out) { try { EntityEvent event = parse(raw); validate(event); out.collect(event); } catch (JsonParseException | MappingException e) { route(ctx, raw, DlqTier.SCHEMA_INVALID, e); } catch (BusinessValidationException e) { route(ctx, raw, DlqTier.BUSINESS_RULE, e); } catch (Exception e) { route(ctx, raw, DlqTier.UNKNOWN, e); } } private void route(Context ctx, String raw, DlqTier tier, Exception e) { OutputTag<DeadLetterRecord> tag = getTierTag(tier); ctx.output(tag, new DeadLetterRecord( raw, tier.name(), e.getMessage(), getStackTrace(e), System.currentTimeMillis(), "", -1L )); } } Define One Output Tag Per Tier Java public static final OutputTag<DeadLetterRecord> DLQ_SCHEMA = new OutputTag<>("dlq-schema-invalid") {}; public static final OutputTag<DeadLetterRecord> DLQ_BUSINESS = new OutputTag<>("dlq-business-rule") {}; public static final OutputTag<DeadLetterRecord> DLQ_UNKNOWN = new OutputTag<>("dlq-unknown") {}; Sink Each Tier Independently Java SingleOutputStreamOperator<EntityEvent> processed = kafkaSource.process(new TieredDlqRouter()); processed.getSideOutput(DLQ_SCHEMA) .sinkTo(buildKafkaSink("dlq.schema-invalid")); processed.getSideOutput(DLQ_BUSINESS) .sinkTo(buildKafkaSink("dlq.business-rule")); processed.getSideOutput(DLQ_UNKNOWN) .sinkTo(buildKafkaSink("dlq.unknown")); This makes the DLQ operationally useful instead of just technically correct. For example: Schema failures can be routed to the producer teamBusiness rule failures can feed data quality workflowsUnknown failures can trigger higher-severity alerting Pattern 4: Choose DLQ Sinks Based on How You Plan To Recover Once records are routed to a DLQ stream, they need a durable destination. In practice, the two most common choices are Kafka and object storage. Kafka DLQ Sink Kafka is the right choice when you want: Near-real-time inspectionStreaming replayOperational integration with existing consumers Java private static KafkaSink<DeadLetterRecord> buildDlqKafkaSink( String topicName) { return KafkaSink.<DeadLetterRecord>builder() .setBootstrapServers("kafka-broker:9092") .setRecordSerializer( KafkaRecordSerializationSchema.builder() .setTopic(topicName) .setValueSerializationSchema( new JsonSerializationSchema<>(DeadLetterRecord.class)) .setKeySerializationSchema( record -> record.errorType().getBytes()) .build() ) .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) .build(); } S3 DLQ Sink Object storage is the better choice when you want: Long retentionLow-cost quarantineBatch replay with Spark or AthenaPartitioned storage by date or error type Java private static FileSink<DeadLetterRecord> buildS3DlqSink() { return FileSink .forRowFormat( new Path("s3://your-bucket/dlq/entity-resolution/"), new JsonRowEncoder<>(DeadLetterRecord.class) ) .withRollingPolicy( DefaultRollingPolicy.builder() .withRolloverInterval(Duration.ofMinutes(15)) .withInactivityInterval(Duration.ofMinutes(5)) .withMaxPartSize(MemorySize.ofMebiBytes(128)) .build() ) .withBucketAssigner( new DateTimeBucketAssigner<>( "error-type='unknown'/year=yyyy/month=MM/day=dd/hour=HH") ) .build(); } A practical production pattern is to use: Kafka for short-term operational handlingS3 for long-term quarantine and replay That gives you both fast response and durable history. Pattern 5: Monitor DLQ Rate, Not Just Job Uptime A DLQ that nobody watches is just a backlog with better branding. Job uptime alone is not enough. A Flink job can stay green while quietly routing 10% of traffic to the DLQ. That is still a production incident. Add Metrics Inside the Operator Java public class MonitoredEntityEventProcessor extends ProcessFunction<String, EntityEvent> { private transient Counter dlqCounter; private transient Counter successCounter; private transient Histogram processingLatency; @Override public void open(Configuration parameters) { MetricGroup metrics = getRuntimeContext() .getMetricGroup() .addGroup("entity_resolution"); dlqCounter = metrics.counter("dlq_routed_total"); successCounter = metrics.counter("processed_success_total"); processingLatency = metrics.histogram( "processing_latency_ms", new DescriptiveStatisticsHistogram(1000) ); } @Override public void processElement( String raw, Context ctx, Collector<EntityEvent> out) { long start = System.currentTimeMillis(); try { EntityEvent event = parseAndValidate(raw); successCounter.inc(); out.collect(event); } catch (Exception e) { dlqCounter.inc(); ctx.output(DLQ_TAG, buildDeadLetter(raw, e)); } finally { processingLatency.update(System.currentTimeMillis() - start); } } } Alert on DLQ Rate A useful alert is DLQ throughput relative to successful throughput: YAML - alert: FlinkDlqRateHigh expr: | rate(flink_entity_resolution_dlq_routed_total[5m]) / rate(flink_entity_resolution_processed_success_total[5m]) > 0.01 for: 2m labels: severity: warning annotations: summary: "DLQ rate exceeds 1% of total throughput" description: "Check dlq.unknown Kafka topic for upstream schema changes" As a rule of thumb: above 1% often indicates schema drift or producer issuesabove 5% usually indicates a broader systemic problem The exact thresholds depend on the pipeline, but the principle does not: monitor DLQ rate as a first-class health signal. Pattern 6: Replay With a Dedicated Reprocessing Job A DLQ is only complete when replay is possible. The cleanest design is a separate Flink job that reads from the DLQ topic and routes records back through the main processing logic. Example Replay Job Java public class DlqReprocessingJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<DeadLetterRecord> dlqStream = env .fromSource( buildKafkaSource("dlq.schema-invalid"), WatermarkStrategy.noWatermarks(), "dlq-source" ); DataStream<String> replayStream = dlqStream .filter(r -> r.failedAtEpochMs() >= START_EPOCH && r.failedAtEpochMs() <= END_EPOCH) .map(DeadLetterRecord::rawPayload); SingleOutputStreamOperator<EntityEvent> reprocessed = replayStream.process(new EntityEventProcessor()); reprocessed.sinkTo(buildDownstreamKafkaSink()); reprocessed.getSideOutput(DLQ_TAG) .sinkTo(buildKafkaSink("dlq.permanent-quarantine")); env.execute("DLQ Reprocessing Job"); } } Why Replay Should Be a Separate Job Keeping replay separate from the main pipeline gives you: Independent scalingIndependent schedulingCleaner checkpoint behaviorSafer operational control It also lets you drain backlogs on your own terms: Off-peak hoursReduced parallelismOr maximum parallelism when you need to catch up quickly That separation keeps the main pipeline stable while still making recovery practical. PyFlink Version: Same Pattern, Same Principle If your team uses PyFlink, the same side output pattern applies. Python from pyflink.datastream import StreamExecutionEnvironment from pyflink.datastream.functions import ProcessFunction from pyflink.common.typeinfo import Types from pyflink.datastream.output_tag import OutputTag DLQ_TAG = OutputTag( "dead-letter-queue", Types.ROW_NAMED( ["raw_payload", "error_type", "error_message", "failed_at_ms"], [Types.STRING(), Types.STRING(), Types.STRING(), Types.LONG()] ) ) class EntityEventProcessor(ProcessFunction): def process_element(self, value, ctx): try: event = parse_and_validate(value) yield event except Exception as e: from pyflink.common import Row yield DLQ_TAG, Row( raw_payload=str(value), error_type=type(e).__name__, error_message=str(e), failed_at_ms=int(time.time() * 1000) ) env = StreamExecutionEnvironment.get_execution_environment() source_stream = env.from_source(...) processed = source_stream.process( EntityEventProcessor(), output_type=Types.STRING() ) good_events = processed dead_letters = processed.get_side_output(DLQ_TAG) good_events.sink_to(build_downstream_sink()) dead_letters.sink_to(build_dlq_sink()) env.execute("Entity Resolution Pipeline") The syntax changes, but the design principle stays the same: good records continue, bad records are isolated and persisted. Production Checklist Before shipping a Flink pipeline, verify the following: RequirementWhy It MattersRisky operators wrapped in try/catchPrevents restart loops from unhandled exceptionsDLQ output tags use explicit typingAvoids runtime serialization failuresDLQ sink is durableFailed records must survive restartsDLQ metrics are exportedSilent DLQ growth is otherwise invisibleReplay path exists and is testedA DLQ without replay is just storageDLQ retention is long enoughTeams need time to diagnose and replayPermanent quarantine existsPrevents infinite replay loopsAlerting is based on DLQ rateJob health alone is not enough This checklist is worth automating in code review or deployment readiness checks. DLQ handling is too important to leave to convention. Key Takeaways If you are building Flink pipelines in production, the safest default is: Use side outputs for DLQ routingRetry transient failures before escalationClassify failures into separate DLQ streamsSink DLQ records durablyExport DLQ metricsReplay through a dedicated job The core rule is simple: A bad message should never silently disappear, and it should never silently stop the stream. That is what turns DLQ handling from a defensive coding trick into a real reliability pattern. Environment Notes The examples in this article target: Apache Flink 1.18Java 17PyFlink 1.18 A few implementation notes: The retry timer pattern requires a keyed stream before KeyedProcessFunctionRocksDB is usually the safer state backend for larger retry stateHashMap state backend can work well for smaller, latency-sensitive workloadsAT_LEAST_ONCE is usually sufficient for DLQ sinks Final Thoughts Poison messages are not rare in streaming systems. They are inevitable. The real question is whether one bad record can take down an otherwise healthy pipeline. With the right DLQ design in Flink, the answer becomes no. The stream keeps moving. Good records continue. Bad records are quarantined. Alerts fire. Replay remains possible. And the pipeline stays operational while the root cause is fixed. That is the difference between a stream that works in staging and one that survives production.

By Rohit Muthyala

Apache Spark Query Optimization on Databricks: Catalyst, AQE, and Photon Engine

Why Query Optimization Matters A Spark query written by a human and a Spark query executed by the engine are often very different things. The gap between them — the optimization — is what separates a job that runs in 3 minutes from one that runs in 3 hours on identical hardware. Databricks compounds Spark's native Catalyst optimizer with two additional layers: Adaptive Query Execution (AQE) – re-optimizes the query at runtime using actual statistics collected mid-jobPhoton – a C++ vectorized execution engine that replaces the JVM-based Spark executor for eligible operators Understanding all three lets you write queries that cooperate with the engine rather than fight it. The Catalyst Optimizer Pipeline Catalyst is Spark's rule-based and cost-based query optimizer. Every query — whether written in SQL, DataFrame API, or Dataset API — passes through the same four-stage pipeline before a single byte of data is read. Stage 1: Parsing — From SQL to Unresolved Logical Plan Python # ── Catalyst Stage 1: Parsing ───────────────────────────────────────────────── # Spark uses ANTLR4 to parse SQL into an Abstract Syntax Tree (AST). # At this point column names are NOT validated — the plan is "unresolved". from pyspark.sql import SparkSession spark = SparkSession.builder.appName("catalyst-demo").getOrCreate() # Both of these produce identical internal representations df_api = ( spark.table("prod.silver.events_clean") .filter("event_type = 'purchase'") .groupBy("platform") .agg({"revenue": "sum"}) ) sql_api = spark.sql(""" SELECT platform, SUM(revenue) AS total_revenue FROM prod.silver.events_clean WHERE event_type = 'purchase' GROUP BY platform """) # Inspect the unresolved logical plan (before analysis) df_api.explain(mode="formatted") # Output includes: # == Parsed Logical Plan == # 'Aggregate ['platform], ['platform, unresolvedAlias('sum('revenue), None)] # +- 'Filter ('event_type = 'purchase) # +- 'UnresolvedRelation [prod, silver, events_clean] The key insight here: UnresolvedRelation and unresolvedAlias mean Spark hasn't touched the catalog yet. Column names could be typos at this point and Catalyst doesn't know. Stage 2: Analysis — Binding to the Catalog The Analyzer walks the unresolved AST and looks up every relation and attribute against the Catalog (in Databricks, this is Unity Catalog). It resolves column names, infers data types, validates references, and binds functions. Python # ── Catalyst Stage 2: Analysis ──────────────────────────────────────────────── # After analysis, every column is resolved to a specific attribute with a type. # AnalysisException is thrown HERE if a column doesn't exist. from pyspark.sql import functions as F from pyspark.sql.utils import AnalysisException # Example of what Analysis catches: try: spark.table("prod.silver.events_clean") \ .select("nonexistent_column") \ .show() except AnalysisException as e: print(f"Analysis failed: {e}") # → AnalysisException: [UNRESOLVED_COLUMN.WITH_SUGGESTION] # A column or function parameter with name `nonexistent_column` cannot be resolved. # After successful analysis, inspect the resolved plan df = ( spark.table("prod.silver.events_clean") .filter(F.col("event_type") == "purchase") .select("platform", "revenue", "user_id") ) # The analyzed plan shows fully qualified attribute IDs like: # == Analyzed Logical Plan == # platform: string, revenue: double, user_id: string # Project [platform#42, revenue#67, user_id#31] # +- Filter (event_type#39 = purchase) # +- Relation prod.silver.events_clean[...] parquet print(df._jdf.queryExecution().analyzed()) Stage 3: Logical Optimization — Rule-Based Rewrites This is where Catalyst applies its ~100+ built-in rules to produce an equivalent but cheaper logical plan. Rules fire repeatedly in fixed-point iteration until the plan stabilises. Python # ── Catalyst Stage 3: Key Optimization Rules ────────────────────────────────── # RULE 1: Predicate Pushdown # Catalyst moves filters as close to the data source as possible, # so Spark reads fewer rows from Parquet. df_before = ( spark.table("prod.silver.events_clean") .join( spark.table("prod.silver.users_clean"), on="user_id" ) .filter(F.col("event_type") == "purchase") # ← filter AFTER join ) # Catalyst rewrites this internally as if you wrote: df_after_equivalent = ( spark.table("prod.silver.events_clean") .filter(F.col("event_type") == "purchase") # ← filter BEFORE join .join( spark.table("prod.silver.users_clean"), on="user_id" ) ) # Result: potentially millions fewer rows shuffled during the join # RULE 2: Column Pruning # Catalyst removes columns not needed by downstream operators. # Even if you select(*), Spark will only read the columns it needs. df_pruned = ( spark.table("prod.silver.events_clean") .select("*") .filter(F.col("event_type") == "purchase") .groupBy("platform") .agg(F.sum("revenue").alias("total_revenue")) ) # Internally, Catalyst prunes all columns except: event_type, platform, revenue # RULE 3: Constant Folding # Expressions with only literals are evaluated at plan time, not per-row. df_constants = spark.range(1000).select( F.lit(2 + 3 * 4).alias("always_14"), # folded to Literal(14) at plan time F.col("id") * F.lit(1).alias("same_id"), # simplified to just col("id") ) # RULE 4: Boolean Simplification # AND/OR chains with tautologies or contradictions are collapsed df_simplified = spark.range(100).filter( (F.col("id") > 10) & F.lit(True) # simplified to just (col("id") > 10) ) # See all optimizations applied: print(df_pruned._jdf.queryExecution().optimizedPlan()) Stage 4: Physical Planning — Strategies and Cost Models The physical planner maps each logical operator to one or more physical implementations and selects the best one using a cost model. The most impactful decision here is join strategy selection. Python # ── Catalyst Stage 4: Physical Planning & Join Strategies ──────────────────── # JOIN STRATEGY 1: Broadcast Hash Join (BHJ) # Best when one side is small enough to fit in executor memory. # No shuffle — the small table is broadcast to all workers. spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10mb") # default large_df = spark.table("prod.silver.events_clean") # 500GB small_df = spark.table("prod.gold.product_catalog") # 8MB ← will be broadcast result_bhj = large_df.join(small_df, on="product_id") # BHJ auto-selected # Force BHJ with a broadcast hint (overrides threshold check): from pyspark.sql.functions import broadcast result_forced = large_df.join(broadcast(small_df), on="product_id") # JOIN STRATEGY 2: Sort Merge Join (SMJ) # Default for large-large joins. Both sides are sorted and merged. # Requires a full shuffle — expensive but handles any size. spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1") # disable BHJ large_df2 = spark.table("prod.silver.transactions_clean") # 200GB result_smj = large_df.join(large_df2, on="user_id") # SMJ selected # JOIN STRATEGY 3: Shuffle Hash Join (SHJ) # Hash-based, no sort. Chosen by AQE when one side is much smaller # than the other but still above the broadcast threshold. spark.conf.set("spark.sql.join.preferSortMergeJoin", "false") # WHOLE-STAGE CODEGEN: Spark fuses multiple operators into a single # Java function to avoid virtual dispatch overhead and intermediate objects. # Verify it's active in your plan: spark.conf.set("spark.sql.codegen.wholeStage", "true") # default result_bhj.explain(mode="formatted") # Look for: *(1) BroadcastHashJoin — the *(N) prefix = WholeStageCodegen stage N Adaptive Query Execution (AQE) AQE is Databricks' most impactful runtime optimization layer. It materializes shuffle map output statistics at shuffle boundaries and uses them to make three key decisions after data has been partially processed. Python # ── AQE Configuration ───────────────────────────────────────────────────────── # AQE is ON by default in Databricks Runtime 7.3+ spark.conf.set("spark.sql.adaptive.enabled", "true") # 1. Dynamic Partition Coalescing # Merges small post-shuffle partitions to avoid thousands of tiny tasks spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128mb") spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionNum", "1") # 2. Dynamic Join Strategy Switching # Allows AQE to downgrade SMJ → BHJ at runtime if a side turns out small spark.conf.set("spark.sql.adaptive.localShuffleReader.enabled", "true") # AQE broadcast threshold (can be higher than static threshold since # we now KNOW the actual size) spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "30mb") # 3. Skew Join Optimization # Splits oversized partitions and replicates the non-skewed side spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true") spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5") # 5x median spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256mb") # Verify AQE decisions in the query plan: df = ( spark.table("prod.silver.events_clean") .join(spark.table("prod.silver.users_clean"), on="user_id") .groupBy("platform") .agg(F.sum("revenue").alias("total")) ) df.explain(mode="formatted") # Look for: AdaptiveSparkPlan isFinalPlan=true # and: == Final Physical Plan == (shows post-AQE decisions) The Photon Engine Photon is Databricks' native vectorized query engine written in C++. It replaces the JVM-based Spark executor for eligible operations, processing data in column-oriented batches (vectors) rather than row-by-row. Python # ── Photon Configuration & Verification ─────────────────────────────────────── # Photon is available on Databricks Runtime 9.1+ with Photon-enabled clusters. # Enable it at the cluster level (UI: Cluster > Configuration > Enable Photon) # or via config: spark.conf.set("spark.databricks.photon.enabled", "true") # Photon-accelerated operators (as of DBR 13.x): # ✅ Scan (Parquet, Delta) ✅ Filter / Project # ✅ Hash Aggregate ✅ Sort # ✅ Broadcast Hash Join ✅ Sort Merge Join # ✅ Window functions ✅ Union / Expand # ✅ String functions ✅ Math functions # ❌ UDFs (Python/Scala) ❌ Some complex types # ❌ Streaming (partial) ❌ RDD-based operations # Verify Photon is executing your query: df = spark.sql(""" SELECT platform, DATE_TRUNC('month', event_ts) AS month, SUM(revenue) AS total_revenue, COUNT(DISTINCT user_id) AS unique_buyers, AVG(revenue) AS avg_order_value FROM prod.silver.events_clean WHERE event_type = 'purchase' AND event_ts >= '2024-01-01' GROUP BY platform, DATE_TRUNC('month', event_ts) ORDER BY month DESC, total_revenue DESC """) df.explain(mode="formatted") # Look for operators prefixed with "Photon" in the physical plan: # == Physical Plan == # PhotonResultStage # +- PhotonSort [month DESC NULLS LAST, total_revenue DESC NULLS LAST] # +- PhotonShuffleExchangeSink hashpartitioning(platform, month) # +- PhotonGroupingAgg [platform, month], [sum(revenue), count(user_id), avg(revenue)] # +- PhotonFilter (event_type = purchase AND event_ts >= 2024-01-01) # +- PhotonScan parquet prod.silver.events_clean # Photon performance metrics appear in Spark UI under "Photon Metrics": # - Photon scan time # - Photon total compute time # - Rows processed by Photon vs fallback JVM Reading Explain Plans The explain(mode="formatted") output is your primary debugging tool. Here's how to read it efficiently: Python # ── Explain Plan Modes ──────────────────────────────────────────────────────── df = ( spark.table("prod.silver.events_clean") .filter(F.col("event_type") == "purchase") .join(broadcast(spark.table("prod.gold.product_catalog")), on="product_id") .groupBy("platform", "category") .agg( F.sum("revenue").alias("total_revenue"), F.count("*").alias("transaction_count") ) ) # Mode 1: simple (default) — compact tree df.explain() # Mode 2: extended — all 4 plan stages side by side df.explain(mode="extended") # Mode 3: formatted — human-readable with operator details (RECOMMENDED) df.explain(mode="formatted") # Mode 4: cost — includes estimated row counts and sizes (requires ANALYZE TABLE) df.explain(mode="cost") # Mode 5: codegen — shows generated Java code for WholeStageCodegen df.explain(mode="codegen") # ── Key Signals to Look For ─────────────────────────────────────────────────── # ✅ GOOD signs: # *(N) prefix → WholeStageCodegen active (operators fused) # BroadcastHashJoin → small table correctly broadcast, no shuffle # PhotonXxx → Photon accelerating this operator # AdaptiveSparkPlan → AQE is engaged # PartitionFilters → Delta/Parquet file skipping active # PushedFilters → filters pushed to Parquet reader # ❌ WARNING signs: # Exchange (shuffle) → unexpected shuffle (missing broadcast hint?) # SortMergeJoin → large-large join (may need Z-ORDER or AQE tuning) # HashAggregate x2 → partial + final agg = shuffle involved # CartesianProduct → missing join condition! Will OOM on large tables # ObjectHashAggregate → non-codegen path, JVM overhead # GenerateXxx → explode() or similar, can't be fused # ── ANALYZE TABLE: feed statistics to CBO ───────────────────────────────────── # Without stats, Catalyst uses default estimates (1M rows, 8 bytes/col). # Run ANALYZE to give the Cost-Based Optimizer real numbers. spark.sql("ANALYZE TABLE prod.silver.events_clean COMPUTE STATISTICS") spark.sql(""" ANALYZE TABLE prod.silver.events_clean COMPUTE STATISTICS FOR COLUMNS user_id, event_type, platform, revenue """) # Now explain(mode="cost") shows real row counts and sizes Tuning Reference Table A quick-reference guide for the most impactful Spark/Databricks configs, what they control, and when to change them: Config KeyDefaultWhat It ControlsWhen to Tunespark.sql.adaptive.enabledtrueMaster AQE switchKeep on; only disable for debuggingspark.sql.adaptive.advisoryPartitionSizeInBytes64mbTarget post-coalesce partition sizeIncrease to 128mb–256mb for large shufflesspark.sql.adaptive.skewJoin.enabledtrueAQE skew splitKeep on; tune skewedPartitionFactor if neededspark.sql.autoBroadcastJoinThreshold10mbStatic BHJ thresholdIncrease to 50mb–100mb if executor memory allowsspark.sql.adaptive.autoBroadcastJoinThreshold30mbAQE runtime BHJ thresholdIncrease if AQE isn't catching small tablesspark.sql.shuffle.partitions200Default shuffle partition countSet to 8 × num_cores for your clusterspark.sql.files.maxPartitionBytes128mbMax bytes per Parquet read partitionReduce for high-parallelism scansspark.databricks.photon.enabledtruePhoton vectorized engineKeep on; disable only for UDF-heavy jobsspark.sql.codegen.wholeStagetrueWhole-Stage CodeGen fusionKeep on; disable only for debuggingspark.sql.statistics.histogram.enabledfalseColumn histograms for CBOEnable after running ANALYZE TABLEspark.sql.cbo.enabledtrueCost-Based OptimizerKeep on; requires ANALYZE TABLE to be usefulspark.databricks.delta.optimizeWrite.enabledtrueAuto bin-pack write filesKeep on for all Delta writes Key Takeaways Catalyst has four stages: Parse → Analyze → Optimize → Plan. Each stage has a distinct job, and understanding them tells you exactly where to look when a query misbehaves.Predicate pushdown and column pruning are the two most impactful automatic optimizations — they reduce the data volume Spark has to move before any aggregation or join.AQE is not a set-and-forget feature: tune advisoryPartitionSizeInBytes to your actual data sizes, and verify its decisions with explain(mode="formatted") — look for AdaptiveSparkPlan isFinalPlan=true.Photon drops in transparently for most SQL and DataFrame operations. The exceptions are Python UDFs, RDD operations, and some complex types — refactor these away from hot paths.Run ANALYZE TABLE ... COMPUTE STATISTICS FOR COLUMNS on your most-joined tables. The CBO's join ordering and strategy decisions improve dramatically with real statistics vs. default estimates.explain(mode="formatted") is your most important debugging tool — learn to read it before reaching for cluster config changes. References Apache Spark — Catalyst Optimizer (Deep Dive Paper, Armbrust et al., SIGMOD 2015)Databricks — Adaptive Query ExecutionApache Spark Docs — Adaptive Query ExecutionDatabricks — Photon RuntimeDatabricks Blog — Photon: A Fast Query Engine for Lakehouse SystemsDatabricks — Cost-Based OptimizerApache Spark — Performance Tuning GuideDatabricks — Broadcast Join Hints"Photon: A Fast Query Engine for Lakehouse Systems" (Behm et al., SIGMOD 2022)Spark by Examples — Explain Plan Modes

By Jubin Abhishek Soni

CORE

Data Pipeline Observability: Why Your AI Model Fails in Production

The 3:00 AM Incident That Changed Everything It was a Tuesday morning when the alerts started firing. Our recommendation engine, the one that drives 30% of our revenue, had tanked. Accuracy dropped from 94% to 58% overnight. The data science team immediately blamed the model. They started tweaking hyperparameters, re-training on new data, and running diagnostics. Nothing worked. I got pulled into the war room at 3:00 AM. The first thing I asked wasn't "What's wrong with the model?" It was "What changed in the data pipeline?" Turns out, everything. A vendor had pushed a schema change upstream. A field that used to be required became optional. Null values started flowing through our pipeline. Our feature engineering code didn't handle nulls gracefully; it just propagated them downstream. By the time the data reached the model, 40% of our feature vectors were corrupted. The model wasn't broken. The data was. We spent six hours manually rolling back the schema change, re-running the pipeline, and restoring service. The incident report was brutal: "Lack of data validation caught a breaking change too late." That's when I realized we needed observability in our data pipeline, not just in our models. The Problem: Data Quality is Invisible Until It Breaks Here's the uncomfortable truth about data pipelines: they fail silently. Your ETL job completes successfully. Your Spark cluster finishes transformations. Your data warehouse loads without errors. Everything looks green in the monitoring dashboard. But the data itself? Garbage in, garbage out. There are three categories of failures that break AI models in production: Missing Values: A source system stops populating a field. Your pipeline doesn't validate it. The model gets NaN values it never saw during training. Predictions become random noise. Schema Changes: An upstream team adds a new column, renames an existing one, or changes data types. Your pipeline doesn't expect these changes. Either it crashes, or worse, it silently maps data to the wrong columns. Distribution Shifts: The statistical properties of your data change. A field that was always between 0 and 100 suddenly has values of 50,000. Your model's scaling assumptions break. Predictions become nonsensical. None of these show up in traditional infrastructure monitoring. Your CPU is fine. Memory is fine. Network is fine. But your data is on fire. The Solution: Observability at Every Layer I started building a three-layer observability framework using dbt, Great Expectations, and custom validation logic. The goal was simple: catch data quality issues before they reach the model. Layer 1: dbt Tests (The First Line of Defense) dbt tests are your cheapest, fastest way to catch obvious data quality issues. They run after every transformation and fail the entire pipeline if something's wrong. Here's what we implemented: SQL -- models/staging/stg_user_events.yml version: 2 models: - name: stg_user_events columns: - name: user_id tests: - not_null - unique - name: event_timestamp tests: - not_null - dbt_utils.expression_is_true: expression: "event_timestamp <= current_timestamp()" - name: event_value tests: - not_null - dbt_utils.expression_is_true: expression: "event_value > 0" These tests are simple but powerful. They catch: Missing required fields (not_null)Duplicate records (unique)Impossible values (event_timestamp in the future)Out-of-range values (negative prices) We run these tests on every dbt run. If any test fails, the pipeline stops. No data reaches the model. No silent corruption. The beauty of dbt tests is that they're version-controlled, documented, and part of your transformation code. When a schema change happens, you update the test, commit it, and everyone knows what changed. Layer 2: Great Expectations (The Statistical Validator) dbt tests catch structural issues. Great Expectations catches statistical anomalies, the subtle shifts that break models. Here's a real scenario: our user_age column had a distribution of 18-65 for two years. Then one day, we started getting ages of 200, 500, 1000. A data entry bug upstream. dbt tests wouldn't catch this because the values are technically valid integers. But Great Expectations would. Python # great_expectations/expectations/user_events_expectations.py from great_expectations.core.batch import RuntimeBatchRequest from great_expectations.data_context import DataContext context = DataContext() suite = context.create_expectation_suite( expectation_suite_name="user_events_suite", overwrite_existing=True ) validator = context.get_validator( batch_request=RuntimeBatchRequest( datasource_name="my_spark_datasource", data_connector_name="default_runtime_data_connector", data_asset_name="user_events" ), expectation_suite_name="user_events_suite" ) # Expect user_age to be between 18 and 120 validator.expect_column_values_to_be_between( column="user_age", min_value=18, max_value=120 ) # Expect event_value to have a mean between 50 and 200 validator.expect_column_mean_to_be_between( column="event_value", min_value=50, max_value=200 ) # Expect less than 5% missing values in critical columns validator.expect_column_values_to_not_be_null( column="user_id", mostly=0.95 ) # Expect the distribution to match historical patterns validator.expect_column_kl_divergence_from_list( column="event_type", partition_object={"event_type": ["click", "view", "purchase"]}, threshold=0.1 ) validator.save_expectation_suite(discard_failed_expectations=False) Great Expectations runs after dbt tests. It validates: Value ranges (age between 18 and 120)Statistical properties (mean event value between 50 and 200)Null rates (less than 5% missing in critical columns)Distribution shifts (event_type distribution matches historical patterns) If Great Expectations detects an anomaly, it alerts us. We investigate before the data reaches the model. Layer 3: Custom Validation (The Domain Expert) dbt and Great Expectations are generic. Your domain is specific. We added custom validation logic that understands our business. Python # pipelines/validation/custom_validators.py import pandas as pd from datetime import datetime, timedelta def validate_feature_engineering(df: pd.DataFrame) -> dict: """ Custom validation for features before they reach the model. Returns a dict of validation results. """ results = {} # Validate 1: Feature completeness # We need at least 95% of features populated feature_cols = [col for col in df.columns if col.startswith('feature_')] null_rate = df[feature_cols].isnull().sum().sum() / (len(df) * len(feature_cols)) results['feature_completeness'] = { 'passed': null_rate < 0.05, 'null_rate': null_rate, 'threshold': 0.05 } # Validate 2: Feature scaling # After normalization, features should be roughly between -3 and 3 (3 sigma) for col in feature_cols: max_val = df[col].max() min_val = df[col].min() results[f'{col}_scaling'] = { 'passed': max_val < 10 and min_val > -10, 'max': max_val, 'min': min_val } # Validate 3: Temporal consistency # Events should be recent (within last 30 days) if 'event_date' in df.columns: df['event_date'] = pd.to_datetime(df['event_date']) days_old = (datetime.now() - df['event_date'].max()).days results['temporal_freshness'] = { 'passed': days_old < 30, 'days_old': days_old, 'threshold_days': 30 } # Validate 4: Business logic # Revenue should always be positive if 'revenue' in df.columns: negative_revenue = (df['revenue'] < 0).sum() results['business_logic_revenue'] = { 'passed': negative_revenue == 0, 'negative_count': negative_revenue } return results def validate_and_alert(df: pd.DataFrame, validation_results: dict) -> bool: """ Check all validations and alert if any fail. Returns True if all pass, False otherwise. """ all_passed = True for check_name, check_result in validation_results.items(): if not check_result['passed']: all_passed = False print(f"ALERT: {check_name} failed") print(f"Details: {check_result}") # Send to monitoring system (Datadog, New Relic, etc.) # send_alert(check_name, check_result) return all_passed This custom validation runs after Great Expectations. It checks: Feature completeness (95% of features populated)Feature scaling (normalized features in the expected range)Temporal freshness (data is recent)Business logic (revenue is positive) If any check fails, we block the pipeline and alert the team. The Real-World Gotchas We Discovered Gotcha 1: Validation Overhead Running dbt tests, Great Expectations, and custom validation on every pipeline run adds latency. We went from 15-minute runs to 25-minute runs. The trade-off was worth it (catching one data quality issue saved us more time than we lost), but you need to plan for it. Gotcha 2: False Positives Great Expectations' distribution shift detection is sensitive. Legitimate business changes (a marketing campaign causing a spike in user_age distribution) triggered false alerts. We had to tune thresholds carefully and add context to alerts. Gotcha 3: Schema Changes Are Sneaky A vendor added a new column to an upstream table. Our pipeline didn't break; it just ignored the new column. But the data science team expected it. We added schema validation to catch new columns and alert us. Gotcha 4: Null Handling Varies Python treats null as None. SQL treats it as NULL. Spark treats it as null. When data flows between systems, nulls get lost or misinterpreted. We had to standardize null handling across the entire pipeline. The Framework: A Decision Matrix Here's how we decide which validation layer to use: Issue TypeCaught ByExampleActionMissing required fielddbt testsuser_id is nullFail pipeline immediatelyDuplicate recordsdbt testsSame user_id appears twiceFail pipeline immediatelyImpossible valuesdbt testsevent_timestamp in futureFail pipeline immediatelyOut-of-range valuesGreat Expectationsage > 150Alert, investigate, fail if severeDistribution shiftGreat Expectationsevent_value mean changes 50%Alert, investigate, continue if acceptableBusiness logic violationCustom validationrevenue is negativeAlert, investigate, failSchema changeCustom validationNew column added upstreamAlert, investigate, update tests The Results: From Chaos to Confidence After implementing this three-layer framework: Incident reduction: We went from 2-3 data quality incidents per month to 0 in six months.Time to resolution: When issues do occur, we catch them within minutes instead of hours.Model stability: Model accuracy stopped fluctuating. It's now consistently 93-95%.Team confidence: Data scientists trust the data. Engineers trust the pipeline. The best part? We caught the schema change incident before it happened. Great Expectations detected the distribution shift, we investigated, found the upstream change, and coordinated with the vendor team before any data reached production. Getting Started: The Minimal Viable Observability You don't need to implement everything at once. Start here: Week 1: Add dbt tests for not_null and unique on critical columns.Week 1: Add dbt tests for not_null and unique on critical columns.Week 1: Add dbt tests for not_null and unique on critical columns.Week 4: Set up alerting so you're notified when validations fail. That's it. You now have observability in your data pipeline. Conclusion: Observability Saves Models Your AI model isn't failing because it's bad. It's failing because the data feeding it is bad. And you won't know the data is bad until you look. The best models in the world can't save you from garbage data. But good observability can. dbt tests, Great Expectations, and custom validation aren't fun. They don't make it into conference talks. But they'll save your production system at 3:00 AM. Start small. Test early. Validate often.

By Abhilash Rao Mesala

Implementing Asynchronous Communication Between Microservices Using Kafka and Spring Boot

In a microservices system, that tight coupling turns a small hiccup into a cascading slowdown. Thread pools fill, retries amplify traffic, and suddenly your simple request is blocked on half the fleet. My executive summary: asynchronous messaging with Kafka helps systems keep moving when individual components inevitably slow down or fail. It does this by decoupling producers from consumers, absorbing traffic spikes, and allowing services to evolve without tying their availability directly to one another. Code Patterns in Spring Boot With Kafka Spring for Apache Kafka gives me two primitives that feel pleasantly old Spring KafkaTemplate for sending and @KafkaListener for receiving. That template/listener model is intentionally similar to other Spring integration tech, which keeps application code focused on domain logic instead of raw client plumbing. Below is a compact (but production-shaped) pattern: externalized config via @ConfigurationProperties, a service port for publishing, a REST command endpoint, a consumer with a real error strategy (DLT), and a REST error advice. Java // === Messaging config (externalized, type-safe) === @ConfigurationProperties(prefix = "messaging.orders") @Validated record OrdersMessagingProps( @NotBlank String topic, @NotBlank String dltTopic ) {} // === DTO (event contract) === public record OrderCreatedEvent(UUID orderId, UUID userId, BigDecimal total, Instant createdAt) {} // === Service port (keeps domain testable, Kafka swappable) === public interface OrderEventPublisher { void publishOrderCreated(OrderCreatedEvent event); } // === Adapter: Kafka producer === @Component class KafkaOrderEventPublisher implements OrderEventPublisher { private final KafkaTemplate<String, OrderCreatedEvent> template; private final OrdersMessagingProps props; KafkaOrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> template, OrdersMessagingProps props) { this.template = template; this.props = props; } @Override public void publishOrderCreated(OrderCreatedEvent event) { // Keying by orderId keeps per-order ordering and drives partitioning decisions. template.send(props.topic(), event.orderId().toString(), event); } } // === REST command API (synchronous edge, async core) === @RestController @RequestMapping("/v1/orders") class OrdersController { private final OrderService orderService; // domain port OrdersController(OrderService orderService) { this.orderService = orderService; } @PostMapping public ResponseEntity<Map<String, Object>> create(@Valid @RequestBody CreateOrderRequest req) { UUID orderId = orderService.create(req.userId(), req.total()); // persists + publishes event return ResponseEntity.accepted().body(Map.of("orderId", orderId, "status", "ACCEPTED")); } record CreateOrderRequest(@NotNull UUID userId, @NotNull @Positive BigDecimal total) {} } // === Domain service port (implementation can use outbox, transactions, etc.) === public interface OrderService { UUID create(UUID userId, BigDecimal total); } // === Consumer: downstream service reacts to events === @Component class BillingListener { @KafkaListener(topics = "${messaging.orders.topic}", groupId = "${spring.kafka.consumer.group-id}") void onOrderCreated(OrderCreatedEvent event) { // Idempotency belongs here: process-by-key + store processed eventId/orderId to avoid duplicates. // Do work (charge card, create invoice, etc.) } } // === Kafka consumer error handling: retries + DLT === @Configuration class KafkaErrorHandlingConfig { @Bean DefaultErrorHandler defaultErrorHandler(KafkaTemplate<Object, Object> template, OrdersMessagingProps props) { var recoverer = new DeadLetterPublishingRecoverer(template, (rec, ex) -> new TopicPartition(props.dltTopic(), rec.partition())); // Backoff and retry policy are configurable; keep it finite to avoid poison-pill loops. return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3)); } } // === REST error handling (ProblemDetail) === @RestControllerAdvice class ApiErrors { @ExceptionHandler(IllegalArgumentException.class) @ResponseStatus(HttpStatus.BAD_REQUEST) ProblemDetail badRequest(IllegalArgumentException ex) { var pd = ProblemDetail.forStatusAndDetail(HttpStatus.BAD_REQUEST, ex.getMessage()); pd.setTitle("Invalid request"); return pd; } } A few been-burned-before notes on the code above. Spring Kafka’s reference docs are explicit that KafkaTemplate is the convenience wrapper for producing, and DefaultErrorHandler + DeadLetterPublishingRecoverer is a first-class way to route failed records to dead-letter topics after retries. If we want non-blocking retries, Spring Kafka also provides @RetryableTopic, which orchestrates retry topics and a DLT automatically useful when transient failures are common and you want predictable retry delay semantics. Containers and Local Dev With Docker Compose When I’m chasing down event flow bugs, I like local environments that feel like the old days: one command, deterministic startup order, and no mystery dependencies. Docker Compose is still the quickest way to stand up Kafka alongside your services, and Confluent publishes straightforward Docker-based tutorials and compose examples for running Kafka locally. For the service image itself, multi-stage builds are the modern classic compile in a builder stage, and copy the artifact into a slimmer runtime stage. Docker documents multi-stage builds as a way to reduce the final image contents and keep build dependencies out of production. Dockerfile # Multi-stage Dockerfile for a Spring Boot service (orders-service) FROM eclipse-temurin:21-jdk AS build WORKDIR /workspace COPY mvnw pom.xml ./ COPY .mvn .mvn RUN ./mvnw -q -DskipTests dependency:go-offline COPY src src RUN ./mvnw -q -DskipTests package FROM eclipse-temurin:21-jre WORKDIR /app COPY --from=build /workspace/target/*.jar app.jar EXPOSE 8080 ENTRYPOINT ["java","-jar","/app/app.jar"] And here’s a Compose file that wires up Kafka and Schema Registry, plus an example Spring Boot service. The exact image choices are illustrative. Your production choices are unspecified and should reflect your standards and security posture. YAML # compose.yaml (local/dev) services: zookeeper: image: confluentinc/cp-zookeeper:7.6.0 environment: ZOOKEEPER_CLIENT_PORT: 2181 kafka: image: confluentinc/cp-kafka:7.6.0 depends_on: [zookeeper] ports: ["9092:9092"] environment: KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 schema-registry: image: confluentinc/cp-schema-registry:7.6.0 depends_on: [kafka] ports: ["8081:8081"] environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092 orders: build: ./orders-service depends_on: [kafka] ports: ["8080:8080"] environment: SPRING_KAFKA_BOOTSTRAP_SERVERS: kafka:9092 MESSAGING_ORDERS_TOPIC: orders.events MESSAGING_ORDERS_DLTTOPIC: orders.events.dlt SCHEMA_REGISTRY_URL: http://schema-registry:8081 Deploying on Kubernetes or AWS On AWS, the Kafka decision is usually managed or self-managed. If you choose Amazon MSK, the cluster lives in your VPC, pick subnets across distinct Availability Zones, and connect clients using the cluster’s bootstrap brokers. That’s the networking baseline, and it’s not optional. MSK is VPC-first by design. For authentication/authorization, MSK supports IAM access control. AWS documents the client configuration for IAM mechanisms. In EKS, I typically pair MSK IAM with IRSA so pods can obtain AWS credentials the AWS way, while ECS services would use task roles instead. Both patterns are documented by AWS, and your choice here is unspecified. Kubernetes service discovery is usually the easy part. Services and Pods get DNS names so workloads can call each other by name rather than IP. Kafka itself is reached via bootstrap broker endpoints or via internal Services, but either way, you want the strings in externalized config, not hardcoded. Here’s a minimal Kubernetes Deployment/Service for a Kafka client service. Values like region, account IDs, and MSK endpoints are unspecified placeholders. YAML apiVersion: apps/v1 kind: Deployment metadata: name: orders namespace: apps spec: replicas: 2 selector: matchLabels: { app: orders } template: metadata: labels: { app: orders } spec: serviceAccountName: orders-sa # IRSA-bound (role ARN unspecified) containers: - name: orders image: <UNSPECIFIED_AWS_ACCOUNT_ID>.dkr.ecr.<UNSPECIFIED_REGION>.amazonaws.com/orders:<TAG> ports: [{ containerPort: 8080 }] env: - name: SPRING_KAFKA_BOOTSTRAP_SERVERS value: "<UNSPECIFIED_MSK_BOOTSTRAP_BROKERS>" - name: MESSAGING_ORDERS_TOPIC value: "orders.events" - name: MESSAGING_ORDERS_DLTTOPIC value: "orders.events.dlt" readinessProbe: httpGet: { path: /actuator/health/readiness, port: 8080 } initialDelaySeconds: 10 --- apiVersion: v1 kind: Service metadata: name: orders namespace: apps spec: selector: { app: orders } ports: - port: 80 targetPort: 8080 Operationally, MSK exposes metrics into CloudWatch (AWS/Kafka), and broker logs can be delivered to CloudWatch Logs (or S3/Firehose). That combination gives you the classic visibility loop: throughput, lag, under-replicated partitions, and error logs without running your own monitoring plane. For distributed tracing in async flows, OpenTelemetry is my default vocabulary now. Spring Boot supports OpenTelemetry export via OTLP, and OpenTelemetry defines Kafka semantic conventions so your producer/consumer spans and attributes stay consistent across tools. CI/CD and the Hard-Earned Field Notes For CI/CD, I keep it boring: build once, push an immutable image, deploy via a declarative mechanism. AWS Prescriptive Guidance provides a clear GitHub Actions pattern for building Docker images and pushing to Amazon ECR, which is a solid baseline when your region/account is unspecified until configured. YAML # .github/workflows/orders.yml name: orders on: push: branches: ["main"] jobs: build_push_deploy: runs-on: ubuntu-latest permissions: id-token: write contents: read steps: - uses: actions/checkout@v4 - uses: actions/setup-java@v4 with: distribution: temurin java-version: "21" - name: Build & test run: ./mvnw -q test package - name: Configure AWS credentials (OIDC) uses: aws-actions/configure-aws-credentials@v4 with: role-to-assume: arn:aws:iam::<UNSPECIFIED_AWS_ACCOUNT_ID>:role/<UNSPECIFIED_GHA_ROLE> aws-region: <UNSPECIFIED_REGION> - name: Login to ECR run: | aws ecr get-login-password --region <UNSPECIFIED_REGION> \ | docker login --username AWS --password-stdin <UNSPECIFIED_AWS_ACCOUNT_ID>.dkr.ecr.<UNSPECIFIED_REGION>.amazonaws.com - name: Build & push image run: | IMAGE=<UNSPECIFIED_AWS_ACCOUNT_ID>.dkr.ecr.<UNSPECIFIED_REGION>.amazonaws.com/orders:${{ github.sha } docker build -t $IMAGE ./orders-service docker push $IMAGE - name: Deploy to EKS (example) run: | aws eks update-kubeconfig --name <UNSPECIFIED_EKS_CLUSTER> --region <UNSPECIFIED_REGION> kubectl -n apps set image deploy/orders orders=$IMAGE Now, the part I wish someone had handed me in 2016: Kafka gives you strong tools, but it does not remove distributed-systems truths. You still need safeguards on the consumer side: idempotent processing, disciplined schema management, and clearly defined retry and dead-letter topic behavior. Kafka’s documentation is careful about the limits of “exactly once” guarantees. Idempotent producers and transactions can strengthen delivery semantics, but achieving true end-to-end exactly-once behavior, especially when external side effects are involved, still depends on deliberate system design. For schema governance, Kafka itself doesn’t ship a schema registry, but acknowledges third-party registries; in practice, Confluent Schema Registry and Apicurio Registry are common choices. Both store schemas out-of-band, so messages carry only a schema identifier, and both support evolvable contracts across Avro/JSON Schema/Protobuf depending on your ecosystem. Conclusion and Best Practices If you take one lesson from my legacy brain into modern event-driven systems, let it be this: asynchrony is a reliability feature, not a performance trick. Kafka’s durable log and consumer group model decouples uptime and absorbs spikes, but you only get the real benefit when you treat schemas as contracts, consumers as idempotent processors, and failure handling as first-class application behavior. On AWS, the operational baseline is non-negotiable. MSK lives in your VPC across AZ subnets, clients connect via bootstrap brokers, IAM auth is configured explicitly, and observability lives in CloudWatch. Do those fundamentals early, and Kafka stops feeling like a mysterious black box and starts feeling like the dependable workhorse it was built to be.

By Mallikharjuna Manepalli

Data Governance Checklist for AI-Driven Systems

Editor’s Note: The following is an article written for and published in DZone’s 2026 Trend Report, Cognitive Databases, Intelligent Data: Unified Infrastructure for Vector Search, AI-Optimized Queries, and Hybrid Workloads. Many teams find governance gaps only after a retrieval system surfaces stale or unauthorized content in production. Models, agents, and retrieval workflows all depend on enterprise data. Before any of that data reaches an AI system, teams need to know where it originates, how it’s integrated, whether it meets quality expectations, what context enriches it, who can access it, and how it changes over time. This checklist gives engineering, data, platform, architecture, and governance teams a structured way to check whether enterprise data is ready for AI use. It focuses on data lifecycle readiness, not model selection or prompt engineering. Use it before production, then revisit the checks during recurring reviews. Table: Data Lifecycle Overview Lifecycle StageWhat to confirmexample evidenceSource readinessOwned, approved, refreshed, understood data sourcesSource catalog entry, owner recordData preparationReliable integration, quality, standardization, enrichmentQuality report, transformation testGovernance continuityClassification, access, lineage, change controlsAccess policy, lineage recordAI-facing assetsDerived assets tied to source rulesDerived asset inventory, retrieval testProduction feedbackMonitoring, issue routing, remediation closureMonitoring alert, remediation log Source Inventory and Ownership AI data governance starts before any source is exposed to an AI system. Teams need to know which sources are in scope, where the data comes from, how often it changes, and who owns its accuracy; being connected to a source is not the same as being approved to use it. Catalog every data source connected to AI environments, including whether it is approved for AI useRequire domain-owner sign-off before approving a connected source for AI workloads; record approval alongside the source entryDesignate the authoritative source for each business entity before its data is copied or exposed for AI useAssign a named domain owner for each source, responsible for accuracy, freshness, and documented limitationsRecord each source’s refresh schedule and acceptable lag; flag sources without a defined scheduleDocument known data gaps, coverage limits, and quality issues at the source level so consuming teams can account for them Integration, Quality, and Enrichment Raw data should not feed AI systems until teams have checked its quality, resolved inconsistencies, and added the business context needed to interpret it correctly. A connected source can still be too coarse, narrow in scope, or out of date for the workflow it feeds. Teams should resolve these mismatches before the data is exposed to AI systems. Validate that integration jobs handle schema changes, missing fields, and source outages without dropping data silentlyDefine measurable quality thresholds (e.g., completeness, timeliness) before a dataset is approvedAssign a team that must resolve quality failures before the data is approvedStandardize formats, naming conventions, and reference values before data enters AI-facing stores, tools, or servicesEnrich records with business context (e.g., department codes, product hierarchies) that downstream systems need to interpret them correctlyDocument the reference datasets and lookups used to enrich AI-facing records so teams can trace added context back to its sourceTest transformations against known inputs and outputs after each change to confirm that business rules still holdReject or quarantine records that fall below quality thresholds before they affect retrieval results or generated responses Classification, Access, and Use Boundaries AI systems should follow least privilege, only using data approved for the user, workflow, and output at hand. The same access rules apply at every stage the data passes through, including storage, indexes, embeddings, retrieval results, caches, and logs. Sensitivity enforced at the source must stay enforced after the data is copied, transformed, or indexed. Classify data assets by sensitivity level and map each level to permitted usesEnforce least-privilege access across source systems, pipelines, indexes, retrieval tools, and AI services so downstream AI use doesn’t bypass source permissionsDocument whether each AI-facing data store, index, or retrieval service inherits source access at query time or enforces copied ACLsMask or remove sensitive fields before they reach AI services, tools, or promptsMaintain approved and prohibited uses for each sensitivity levelSeparate dev, staging, and prod environments so live data does not leak into experimental systemsRequire explicit approval before adding a new data source or sensitivity category to an AI system Lineage, Provenance, and Change Traceability When a model or agent produces an unexpected result, teams need to trace the data from source to output, with enough detail to link a specific AI response to the inputs behind it. The same trail supports audit and regulatory reviews. Without it, a team investigating an issue has to guess whether the cause was a stale source, broken transformation, or out-of-date index. Capture the source system, extraction time, transformation version, and pipeline run ID for each record prepared for AI useTrack schema changes, business rule updates, and definition/version changes for fields that affect AI interpretation (e.g., “active customer”)Maintain provenance metadata for enrichment steps so added business context can be traced to its sourceLink derived assets (e.g., embeddings, indexes, summaries) to the source records and pipeline versions that produced themRetain lineage records for the period required by regulatory and audit policiesStore lineage records in a system queryable by data, platform, and audit teams independently of the pipelines that produced them Embeddings, Indexes, and Derived Data Assets Embeddings, indexes, summaries, and caches are copies of source data shaped for retrieval, so ownership, classification, access, and lineage controls must carry forward. When a copy falls out of sync with its source, AI systems may retrieve stale context or keep information that should have been updated or deleted. Assign an owner accountable for the accuracy and freshness of each embedding store, vector index, summary cache, or other derived assetDefine a refresh cadence that keeps each derived asset aligned with source data within a documented latency toleranceVersion-derived assets so teams can roll back after a bad source change or failed updateApply the same source system retention, deletion, and access policy rules and changes to derived assetsValidate index, embedding, summary, and cache updates to confirm they return expected results without dropping recordsLog each derived asset creation, update, and deletion with enough detail to link the change to a specific pipeline run AI-Facing Delivery and Retrieval Reliability Upstream governance only matters if the right information reaches the model or agent when it is needed. Retrieval quality problems are usually data quality problems in another form: Stale sources and lagging indexes can both produce confidently wrong answers. Define retrieval quality expectations, including relevance, freshness, and source attribution, for each AI-facing service or tool; assign a named owner accountable for the specDefine when retrieval should return an answer, return search results only, ask for clarification, or return no answerRequire source attribution for retrieval results that cite internal policies, contracts, customer records, account records, or regulated content so generated responses can be checked against the original dataSet latency and throughput targets for retrieval services so slow or overloaded systems do not degrade model responses or agent actionsConfigure alerts when retrieval quality, freshness, or latency falls below thresholds that could affect retrieval results, generated responses, or agent actionsRequire human review for AI-generated outputs that authorize actions, commit transactions, or affect regulated decisionsTest services and tools end to end with representative queries to confirm that responses use the expected sources Monitoring, Feedback, and Lifecycle Change Production reviews should catch stale data, delayed refreshes, quality drift, and unusual access patterns before they affect AI behavior. Recurring AI output issues should be traced to a specific data source, pipeline step, or derived asset so teams can fix the underlying cause. Flag datasets that miss the refresh window defined for their sourceTrack lag between source updates and derived asset refreshes to detect stale responsesConfigure alerts for unusual access patterns (e.g., unapproved users, services, or tools)Assign recurring AI output issues to the responsible data source, pipeline step, or derived asset owner; record the remediation and closureDefine a deprecation process that identifies which pipelines, services, and derived assets must be updated or retired when a source is removedRequire rollback procedures for source changes, schema migrations, and derived asset updates that could degrade AI behaviorConduct recurring reviews to confirm governance controls still match current use cases and access patterns Closing Data readiness for AI is not a one-time launch task. Build these checks into existing data quality and platform reviews, then revisit them when sources, access rules, derived assets, or AI use cases change. This is an excerpt from DZone’s 2026 Trend Report, Cognitive Databases, Intelligent Data: Unified Infrastructure for Vector Search, AI-Optimized Queries, and Hybrid Workloads.Read the Free Report

By Abhishek Gupta

CORE

The Real-Time Revolution: Why Blockchain Needs Data Stream Processing

Blockchain is an extremely data-driven technology because its primary function is to store, verify, and coordinate independent records in a secure, distributed data network. Without this information, no transaction, smart contract execution, or network activity would be valid, and it could jeopardize the integrity of much larger functions of trust. The data coming into the blockchain affects the accuracy of the whole system. Blockchain is nothing without the data it connects to, so, as far as transparency, immutability, and safe decisions are concerned, data is the backbone of blockchain. Blockchain and data streaming are bringing unprecedented levels of security, transparency, and real-time mechanisms to move data across the digital world. Blockchain forms an unbreakable chain of trust through keeping decentralized records, and streaming data streamlines the process by allowing for insights when information is constantly flowing. These form the backbone of next-generation applications, unleashing innovation, scalability, and better decision-making across industries. Both blockchain and data streaming are independently large and powerful technologies as they exist in the present time. However, when combined, data streaming can amplify the potential impact of a blockchain solution. Real-Time Data Integration Data streaming platforms, such as Apache Kafka and Apache Flink, continuously process and deliver real-time data. When we integrate with blockchain, transactions can be updated instantly on the ledger, smart contracts can react to live data feeds, and delays can be reduced compared to batch processing. For example, we can visualize it as the IoT sensors streaming temperature data can trigger a blockchain-based smart contract in real time. Improved Scalability One major limitation of blockchain systems like Ethereum has been scalability. By leveraging data streaming, we can pre-process and filter large volumes of data before sending it to the blockchain. Can reduce unnecessary transactions that are stored on-chain, and, on top of that, offload heavy computation on-chain and push it to a stream processing engine that is available on data streaming platforms.This results in faster and more efficient blockchain performance. Enhanced Data Integrity and Trust Blockchain ensures immutability and transparency; on the other hand, data streaming ensures continuous data flow. As data stream processing enables continuous validation, filtering, and analysis of data elements before they are processed on the ledger, it enhances data integrity and trust in blockchain. Real-time processing helps identify anomalies in the data, prevent tampering, and ensure that only accurate, high-quality data enters the blockchain. Combining this provides a trusted, secure, and eventually transparent ecosystem in which information can be verified instantly and with confidence. We can consider a use case of supply chain tracking where real-time shipment data is streamed and permanently recorded. Better Event-Driven Architectures Blockchain systems can become more dynamic when combined with an event-driven streaming platform such as Confluent, Amazon Kinesis, or the open-source Apache Kafka. Smart contracts can act as automated responders to streamed events and can be enabled for automation across distributed systems, which finally reduces manual intervention. For example, a payment is automatically released when a delivery event is streamed and confirmed. Efficient Data Storage Strategy Not all data needs to be stored on-chain, which is expensive and slow, but by leveraging streaming platforms, we can store and process high-volume data off-chain. Streaming platforms can be integrated with streaming databases to store data already processed by stream engines. We can allow the Blockchain to store only critical summaries, hashes, or proofs, maintaining efficiency while ensuring verification. Real-Time Analytics and Monitoring Data stream processing facilitates real-time analytics and monitoring in blockchain by analyzing transaction data as it streams over the network. This enables organizations to detect suspicious activity, monitor system performance, and obtain real-time information on blockchain activity by analyzing transaction patterns. Transparency, responsiveness, and operational efficiency across blockchain ecosystems can be upgraded if we convert the raw data into actionable intelligence by integrating a real-time stream processing platform. Wrapping Up Combining these two technologies — data stream processing and blockchain — creates an ecosystem that blends real-time intelligence with secure, immutable record-keeping. Blockchain ensures transparency, trust, and data integrity, while stream processing powers instant analysis, continuous monitoring, and real-time decision-making based on that data. When combined, they improve power efficiency, enhance security, and enable scalable, data-driven applications. These technologies play an instrumental role in the construction of smarter, more intelligent systems that must respond to increase confidence among organizations relying on that real-time information.

By Gautam Goswami

CORE

Parallel Kafka Batch Processing With Kotlin Coroutines in Spring Boot

Managing high-volume message traffic in distributed architectures is crucial. Efficient use of database and CPU resources is also very important. There are structures that allow us to receive messages in batches. The default Spring Kafka "BatchMessageListener" structure addresses this need. However, the processing of these messages often goes through a sequential bottleneck. This article will discuss the structure and usage of Kotlin Coroutines in detail. We will examine how to maximize Kafka message processing performance using Structured Concurrency principles and Resource Throttling techniques. Architectural Bottleneck: Sequential I/O Blocking On the current Kafka listener: Database or external service calls made for each message directly increase total processing times. If the processing speed of a message lags behind the message arrival speed and the max-poll-interval-ms time is exceeded, the consumer is removed from the consumer group. Rebalancing is triggered, and the partitions of that consumer are redistributed to other consumers in the group. Kotlin @KafkaListener(topics = ["usage-pool-topic"]) fun usagePoolListener(records: List<ConsumerRecord<String, String>>) { records.forEach { record -> processRecord(record) // Network latency + DB I/O blocking } } Solution 1. Batch-Fetch and In-Memory Map Structure Before any concurrent code is entered, data is retrieved collectively from all necessary entities. Multiple separate queries are converted into a batch query before data processing begins. The N+1 query problem is solved at the application layer. All data is cached once before being broken down into concurrent operations. Having the data cached significantly reduces our reliance on the database. Using the associateBy function, we transform the data into a map structure with X access times. This allows us to read the data safely from the maps instead of reading each concurrent operation from the database. Kotlin val messages = records.map { objectMapper.readValue(it.value(), UsagePoolRecord::class.java) } val usagePoolEntities = usagePoolRepository .findByIds(messages.map { it.usagePoolId.toBigInteger() }) .associateBy { it.usagePoolId } val lockEntities = lockRepository .findByUserIds(messages.map { it.userId }) .associateBy { it.userId } 2. Structured Concurrency Memory Management With Chunking The chunk structure serves two purposes. It prevents the creation of coroutines simultaneously. This prevents unnecessary memory usage. Each chunk writes to the database after all coroutines have completed their operations. Unnecessary connection pool consumption is avoided. Kotlin messages.chunked(150).forEach { chunk -> // Each chunk of 150 records is processed concurrently } Resource Isolation With limitedParallelism Why limitedParallelism? If the database connection pool has, for example, X connections, keeping the parallelism limit below X prevents "Connection Timeout" errors. Kotlin messages.chunked(150).forEach { chunk -> val deferredResults = chunk.map { record -> CoroutineScope(Dispatchers.IO.limitedParallelism(15)).async { try { processRecord(record, usagePoolEntities, lockEntities) } catch (e: Exception) { log.error("Operation error: ${record.key()}", e) buildErrorRecord(record, e) } } } val results = deferredResults.awaitAll() // Structural waiting collectAndAggregate(results) } The Dispatchers.IO.limitedParallelism(X) command limits the number of concurrent coroutines to X, preventing the DB connection pool from being exhausted.Each coroutine returns a result with the async command. The awaitAll() command waits for all coroutines in the chunk to finish before proceeding to the next step. runBlocking This function blocks callers until all concurrent operations are complete. This is the correct approach here because: It ensures that the Kafka consumer remains blocked to maintain its offset commit structure until all records in the batch are processed. We still benefit from concurrent operation parallelism within the runBlocking block. 3. Thread-Safe Result Structure After the awaitAll() operation, all results are collected in thread-safe queues. Then a single batch write operation takes place. Using MutableList structures to combine results returned from parallel processed coroutines can lead to data loss. At this point, lock-free data structures should be preferred. ConcurrentLinkedQueue uses CAS (Compare-And-Swap) algorithms instead of synchronized blocks. This provides superior performance in high-content write operations. Why Shouldn't We Use ConcurrentLinkedQueue? Concurrent operations (concurrent functions) perform simultaneous write operations to a shared collection of results. Using MutableList leads to race conditions. It performs well in secure and concurrent write operations. Kotlin data class AggregatedRecords( val processedSave: ConcurrentLinkedQueue<ProcessedEntity> = ConcurrentLinkedQueue(), val toDelete: ConcurrentLinkedQueue<UsagePoolEntity> = ConcurrentLinkedQueue(), val retryQueue: ConcurrentLinkedQueue<RetryEntity> = ConcurrentLinkedQueue() ) The DataIntegrityViolationException return is important. When two consumer instances are processing the same record, one of them falls into a unique constraint violation. Instead of making the entire batch fail, record-by-record deletion is performed. Kotlin AggregatedRecords.processedSave .chunked(150) .forEach { batch -> try { processedRepository.saveAll(batch) } catch (e: DataIntegrityViolationException) { batch.forEach { record -> try { processedRepository.save(record) } catch (e: DataIntegrityViolationException) {} } } } 4. Error Tolerance in Write Operations Batch write (saveAll) operations are performant. However, a "Unique Constraint" error in a single record can cause the entire batch to fail. The following structure is critical to meet Optimistic Locking or Idempotency requirements. Kotlin aggregatedRecords.processedSave.chunked(150).forEach { batch -> try { processedRepository.saveAll(batch) } catch (e: DataIntegrityViolationException) { // Fallback: Try one by one if batch fails batch.forEach { record -> try { processedRepository.save(record) } catch (innerException: DataIntegrityViolationException) { log.warn("Duplicate record skipped: ${record.id}") } } } } 5. Data Flow Diagram Ingress: The Kafka batch is caught with runBlocking.Preparation: All necessary context data is retrieved bulk from the DB.Execution: Coroutines are started asynchronously in chunks.Synchronization: The completion of all coroutines is awaited as a barrier point with awaitAll().Egress: Collected results are made permanent with saveAll. Performance Analysis and Results Conclusion Processing Kafka messages in Spring Boot with Kotlin Coroutines not only increases speed but also improves code readability and makes resource management deterministic (predictable). The use of runBlocking allows us to build a bridge between the blocking Kafka consumer thread and the suspended world without disrupting Kafka's offset management mechanism. Dependencies XML <dependency> <groupId>org.jetbrains.kotlinx</groupId> <artifactId>kotlinx-coroutines-core</artifactId> <version>1.7.3</version> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>

By Erkin Karanlık

From ETL to Lakeflow: Shifting to a Declarative Data Paradigm

If you've worked on a data platform for more than a few years, you've almost certainly built the same pipeline twice. First, the way the team wrote pipelines in 2019: a notebook here, a Python script there, an Airflow DAG to glue it all together, and a long document explaining the order things had to run in. Then the rewrite, two years later, when somebody quit, and nobody could remember why a particular task had a sleep(180) in it. Lakeflow is Databricks' answer to that pattern, and the shift it's pushing for is bigger than the marketing makes it sound. It isn't a new orchestrator. It's a move from imperative pipelines, where you write the steps, to declarative pipelines, where you write the destination and let the engine figure out the steps. What follows is the practical version of that shift — what's actually different, where the gains are real, and how to migrate without ending up with a half-converted lakehouse. 1. The Imperative ETL Trap: Why Traditional Pipelines Are Hitting a Wall Imperative ETL is a fancy name for the way most pipelines are still written: a sequence of steps, hand-ordered, run on a schedule. It works fine until it doesn't, and the failure modes are remarkably consistent across teams I've worked with: The DAG outgrows its author. The person who wrote the original 30-task Airflow DAG moves teams. The next engineer is afraid to delete anything because they can't tell which tasks are still needed.Backfills are surgical operations. Re-running yesterday means manually figuring out which downstream tables are stale, in what order. Half the team's tribal knowledge lives in Slack threads about backfills.Quality checks are bolted on. Data quality lives in a separate framework, often a separate codebase, often run by a separate team. By the time a check fails, the bad data is already in the warehouse.Lineage is a slide in a deck. Whatever lineage exists was drawn by hand for a quarterly review and was out of date the day after. None of these are bugs in the imperative model. They're features of it. When you write the steps, you own the steps — including all the cross-task assumptions the engine doesn't know about. 2. What "Declarative" Actually Means in Lakeflow Declarative is one of those words that gets used loosely. In Lakeflow Pipelines, it has a specific, narrow meaning: you describe each table's logical definition (its source query, its expected schema, its quality rules), and the engine determines execution. It picks the order. It decides which tables are streaming and which are batch. It scales the cluster. It figures out incremental processing. It produces lineage automatically because lineage is now a derived property of the dependency graph it built for you. What it isn't: It isn't "low-code." You're still writing SQL or PySpark. The thing that's gone is the orchestration boilerplate around it.It isn't a magic upgrade for any pipeline. Pipelines that genuinely need procedural logic — multi-step API calls with branching, complex pre/post-processing — still belong in Lakeflow Jobs (the orchestrator) or even external code, called from the pipeline.It isn't free. There's a learning curve in stopping yourself from writing the steps you used to write. The first month, most teams over-specify. The mental shift: stop describing how the data should flow. Describe what each table is. Lakeflow figures out the flow. 3. The Lakeflow Architecture: Connect, Pipelines, Jobs Lakeflow is three components that share one governance layer (Unity Catalog). They map roughly onto the three traditional layers of a pipeline — ingestion, transformation, orchestration — but with the imperative wiring removed. Figure 1. Lakeflow's three components on top of Unity Catalog. Pipelines is the declarative core; Connect feeds it, Jobs schedules it. A few practical points about this picture. Lakeflow Connect is where managed connectors live (Salesforce, Workday, Postgres CDC, and a steadily growing list); it's the part you reach for instead of writing yet another ingestion script. Lakeflow Pipelines is where the declarative paradigm actually lives — every other component is conventional. And Lakeflow Jobs is the part that looks most like Airflow: task graphs, retries, alerts. The trick is that the things inside a Pipelines task aren't tasks themselves — they're table definitions, and the engine builds the internal DAG from their dependencies. 4. Translating an Imperative Pipeline to a Declarative One The clearest way to feel the difference is to look at the same logic written both ways. Imagine a small bronze→silver→gold pipeline for transactions: ingest raw files, deduplicate, then aggregate to daily totals. 4a. The imperative version (notebook + Airflow style) Python # bronze.py df = spark.read.json("s3://landing/txns/") df.write.format("delta").mode("append").saveAsTable("bronze.txns") # silver.py -- runs after bronze finishes raw = spark.table("bronze.txns") clean = (raw.dropDuplicates(["txn_id"]) .filter("amount IS NOT NULL")) clean.write.format("delta").mode("overwrite").saveAsTable("silver.txns") # gold.py -- runs after silver finishes agg = (spark.table("silver.txns") .groupBy("ingest_date", "account_id") .sum("amount") .withColumnRenamed("sum(amount)", "daily_total")) agg.write.format("delta").mode("overwrite").saveAsTable("gold.daily_totals") # airflow_dag.py -- the part that actually controls execution bronze_task >> silver_task >> gold_task 4b. The same logic, declared in a Lakeflow Pipeline Python import dlt from pyspark.sql.functions import sum as _sum @dlt.table( name="bronze_txns", comment="Raw transactions landed from S3.", ) def bronze_txns(): return (spark.readStream .format("cloudFiles") .option("cloudFiles.format", "json") .load("s3://landing/txns/")) @dlt.table(name="silver_txns", comment="Deduplicated, validated transactions.") @dlt.expect_or_drop("non_null_amount", "amount IS NOT NULL") @dlt.expect("unique_txn", "txn_id IS NOT NULL") def silver_txns(): return (dlt.read_stream("bronze_txns") .dropDuplicates(["txn_id"])) @dlt.table(name="gold_daily_totals") def gold_daily_totals(): return (dlt.read("silver_txns") .groupBy("ingest_date", "account_id") .agg(_sum("amount").alias("daily_total"))) Two things vanished in the rewrite. There is no DAG file, because the dependencies are inferred from dlt.read / dlt.read_stream calls. There is no separate data quality framework — quality lives next to the table definition, where it belongs. The engine decides what's streaming and what's batch from the calls themselves; bronze is a stream, silver is a stream of the bronze stream, gold is a batch over silver. None of that ordering is in the code I wrote. 5. Quality, Lineage, and Operational Visibility for Free The expectations decorators above (@dlt.expect, @dlt.expect_or_drop, and the stricter @dlt.expect_or_fail) are not just convenience syntax; they become first-class objects in the pipeline. Every run produces a per-expectation pass/fail count, queryable directly: SQL -- How many silver rows failed each expectation, per run, last 7 days SELECT pipeline_run_id, flow_name, expectation_name, passed_records, failed_records, dropped_records FROM event_log("<pipeline-id>") WHERE event_type = 'flow_progress' AND timestamp >= current_timestamp() - INTERVAL 7 DAYS ORDER BY timestamp DESC; Lineage shows up automatically in Unity Catalog — both the table-level edges (gold_daily_totals depends on silver_txns) and column-level edges (gold's daily_total derives from silver's amount). Operationally, this is the change that has the largest day-to-day impact: when somebody asks "what does this column mean and where did it come from," you stop having to guess. What this replaces: Great Expectations runs scheduled separately, OpenLineage stitched together by hand, and a homegrown observability dashboard reading task logs. All three of those projects either go away or shrink dramatically. 6. Migration Strategy: How Teams Actually Move Off Imperative Pipelines I've not seen a successful big-bang migration. The pattern that works is layered: Phase 1 — New pipelines only Make Lakeflow Pipelines the default for any new pipeline. This sounds obvious; the discipline is in saying no when somebody wants to add "just one more" Airflow DAG to the imperative side because it's faster this week. Phase 2 — Convert the painful ones Pick the existing pipelines that hurt the most — the ones with the longest backfill stories, the most ad-hoc quality checks, the worst lineage gaps. Those are the ones where the declarative model pays for the rewrite cost fastest. Don't start with the easy ones; their owners won't thank you for the disruption. Phase 3 — Retire the orchestration boilerplate Once a critical mass of pipelines has moved over, you can shrink (or in many cases delete) Airflow setups, custom dependency-tracking tools, and the side projects that grew up around imperative ETL. This is the phase where the cost savings actually show up in headcount and infrastructure bills. Migration step Effort Watch out for New pipelines on Lakeflow Low Team momentum — easy to revert to old patterns. Convert the top 3 painful pipelines Medium Different streaming/batch semantics in expressed dependencies. Move expectations off external DQ tools Medium Existing alerting wired to the old framework. Retire imperative orchestrator High External callers (BI tools, ML jobs) that triggered DAGs directly. 7. Where Declarative Still Hurts: Honest Limitations I'd be lying if I said this was free. The places where the declarative model still bites: Procedural logic doesn't fit. If your "pipeline" is really a sequence of API calls with branching error handling, that's a Lakeflow Job (or external code), not a declarative table.Cross-pipeline orchestration is its own thing. Lakeflow Pipelines builds the DAG inside a pipeline. If you need pipeline A to wait for pipeline B, you still need Lakeflow Jobs above them.Debugging shifts from steps to definitions. When something is wrong, you're not stepping through a script — you're reading the event log and figuring out which expectation or upstream table caused it. The tooling is good; the muscle memory is different.Cost can surprise you. Auto-scaling on a misbehaving streaming source has the same risk it always has. Set max workers thoughtfully on day one; don't leave it to defaults. Conclusion The shift to declarative pipelines isn't really about syntax. It's about who owns the boring parts. In an imperative pipeline, the team owns the order, the retries, the lineage, the quality checks, and the cluster scaling — and pays in headcount when any of those break. In a declarative pipeline, those become properties of the engine, and the team owns the part that's actually interesting: the table definitions and the business logic. Lakeflow is the cleanest implementation of that idea I've used in production, and the teams I've watched migrate haven't asked to go back.

By Seshendranath Balla Venkata

Rust-Native Alternatives to Spark SQL and DataFrame Workloads

Apache Spark is one of the most powerful tools in the data and AI engineering world. It helps process massive datasets and is widely used across industries, irrespective of cloud platforms. But when you move from learning Spark to running it in production, you start seeing real challenges. This is from practical experience. 1. JVM Overhead Spark runs on the Java Virtual Machine (JVM). At first, this looks fine. But in real workloads, it creates overhead. What actually happens: Extra memory is consumed by the JVM itselfData moves between Python and JVM (serialization)Job startup takes more time Why it matters: Even if your logic is simple, the JVM layer adds hidden cost and latency. Especially in PySpark workloads, this becomes very noticeable. 2. Garbage Collection (GC) Issues The JVM uses garbage collection (GC) to manage memory. In small workloads, no problem. In large workloads, big problem. What we generally observe: Sudden pauses during execution, Jobs becoming slow without a clear reason, and performance behaving inconsistently. Real Challenge We often need to tune: memory settings, GC configuration, and executor behavior. Without proper tuning, performance becomes unpredictable. 3. Cluster Complexity Spark is not just a tool — it is a distributed system. To run it, you must manage infrastructure. What we need to handle: Cluster setup, executors and memory configuration, partition tuning, scaling (up/down). Impact in real projects: Higher infrastructure cost, more operational effort, requires deep expertise, and this adds overhead beyond just writing data pipelines. Rust Changes Everything Rust solves these problems at the language level. No JVM Rust compiles directly to machine code. So, no virtual machine and no runtime overhead. No Garbage Collection Rust uses ownership-based memory management. Memory is handled at compile time No runtime GC pauses Predictable Performance Better memory control, no hidden pauses, Efficient execution Result: Faster and more stable systems When we look at Rust tools, we see different ways: Replace Parts of Spark PolarsDataFrame processingDataFusionSQL engineBallistaDistributed executionRisingWaveStreamingSailFullSpark replacement Lakesail has came up with all together at once place. What Is Sail? Sail is an open-source computation framework that serves as a drop-in replacement for Apache Spark (SQL and DataFrame API) in both single-host and distributed settings. Built in Rust, Sail runs ~4x faster than Spark while reducing hardware costs by 94%. In simple terms: Sail = Spark experience + Rust performance + no JVM/GC problems It is not just a library. It is a full data platform / compute engine. Core Idea of Sail Traditional Spark: Plain Text PySpark → JVM → Spark Engine → Execution Sail: Plain Text PySpark → Spark Connect → Sail (Rust Engine) → Execution Key difference: Spark depends on JVMSail removes the JVM completely Where Sail Is Strong Sail is a good choice if you are already using Apache Spark and want better performance.It allows you to continue using the same Spark SQL and DataFrame APIs without rewriting your code.It removes JVM and garbage collection overhead, which helps improve speed and memory usage.Because it runs on a Rust-native engine, it provides more stable and predictable performance.It can help reduce infrastructure cost while keeping your existing development approach. Where You Should Be Careful Sail is still a new technology and not as mature as the Spark ecosystem.The number of connectors, integrations, and community support is smaller compared to Spark.Some advanced Spark features may not be fully supported yet.It is important to test Sail with your own workload before using it in production. Sail supports almost all modern platforms' emerging features: Local mode (single machine)Cluster mode (Kubernetes) It includes: Task schedulingResource managementDistributed execution Similar to a Spark cluster, but lighter Lakehouse Support Sail supports: Delta LakeApache Iceberg That means: Works with modern data lakesCompatible with existing data Storage Support Sail can read/write from: AWS S3Azure Data LakeGoogle Cloud StorageHDFSLocal files So, it integrates with existing ecosystems Catalog Integration Supports: Unity CatalogIceberg REST Catalog Important for: GovernanceAccess controlEnterprise data management Multimodal + AI Workloads Sail goes beyond Spark. It supports: Structured dataImagesPDFsAI workloads This is called: Multimodal lakehouse. Performance and Cost Sail claims: ~4x faster executionUp to 8x in some workloads~94% lower cost Reasons: No JVM overheadNo GCBetter memory usage Conclusion Sail is a new way to run Spark workloads using Rust instead of the JVM. It removes garbage collection and reduces memory and performance issues, making execution faster and more stable. One of its biggest advantages is that you can keep the same Spark code with little or no changes. This helps reduce infrastructure cost and complexity. However, it is still a new technology and not as mature as Spark yet. In the future, the best approach will be to use the right mix of Spark and Rust tools together.

By Srinivasarao Rayankula

Combining Temporal and Kafka for Resilient Distributed Systems

Kafka and Temporal address different failure boundaries, and resilient distributed systems often need both rather than one as a substitute for the other. Kafka is built to move ordered, replayable event streams across many consumers and machines, while Temporal is built to keep long-running application logic alive as durable Workflow Executions that recover from crashes, outages, and worker restarts by replaying persisted Event History. The combination becomes compelling when Kafka is used to carry facts and Temporal is used to remember intent, timers, retries, and compensations across the lifetime of a business process. Kafka as the Event Backbone and Temporal as the Control Plane Kafka’s model is centered on totally ordered partitions, consumer groups, and offsets. A partition is consumed by exactly one consumer in a subscribing consumer group at a time, and Kafka keeps consumer state compact by treating progress as an offset that can be checkpointed, committed manually, or even rewound for reprocessing. That model is excellent for integration boundaries, stream processing, and decoupling producers from downstream services. What it does not provide by itself is durable orchestration for business logic that must wait for hours, react to multiple messages over time, and recover mid-process without rebuilding state externally. Temporal fills that gap by treating a Workflow Execution as a durable, reliable, scalable function that owns local state, receives messages through Signals or Updates, and advances by replaying persisted history instead of starting over from scratch after failure. Keep Kafka at the Boundary of Workflow Replay The most important design rule is simple: Kafka client calls do not belong inside Workflow code. Temporal requires deterministic workflow logic on replay, and its documentation explicitly places non-deterministic work, such as API calls and database queries, inside Activities. A Workflow should behave like a compact state machine that decides what should happen next, while Activities perform the side effects that may fail or need retries. That separation is what allows Kafka to remain an external event fabric without corrupting Temporal replay semantics. Java private boolean paymentReceived; private final OrderActivities activities = Workflow.newActivityStub( OrderActivities.class, ActivityOptions.newBuilder() .setStartToCloseTimeout(Duration.ofSeconds(30)) .setRetryOptions( RetryOptions.newBuilder() .setInitialInterval(Duration.ofSeconds(1)) .setMaximumInterval(Duration.ofSeconds(30)) .build()) .build()); @WorkflowMethod public void process(String orderId) { activities.reserveInventory(orderId); boolean paid = Workflow.await(Duration.ofHours(2), () -> paymentReceived); if (!paid) { activities.releaseInventory(orderId); activities.publishTimedOut(orderId); return; } activities.publishConfirmed(orderId); } @SignalMethod public void paymentCaptured(String paymentId) { paymentReceived = true; } This workflow is intentionally boring, which is precisely why it is robust. Inventory reservation and event publication are pushed into Activities, while the workflow itself only keeps state and waits. The two-hour wait is not a sleeping thread in application memory; Temporal persists timers so the execution resumes even after worker or service interruptions. Kafka, in this pattern, supplies the external payment event, but Temporal owns the long-lived timeout and the recovery semantics. A thin Kafka bridge can then translate an incoming record into a Temporal message instead of embedding orchestration logic in the consumer loop. Signal-With-Start is especially useful because it either signals an existing workflow or starts a new one with the same Workflow ID and immediately applies the signal, which removes a large class of race conditions between creation and update. Java public void onMessage(ConsumerRecord<String, PaymentEvent> record) { WorkflowStub workflow = client.newUntypedWorkflowStub( "OrderWorkflow", WorkflowOptions.newBuilder() .setWorkflowId("order-" + record.key()) .setTaskQueue("order-workflows") .build()); workflow.signalWithStart( "paymentCaptured", new Object[] { record.value().paymentId() }, new Object[] { record.key() }); consumer.commitSync(); } That handoff should be designed as duplicate-tolerant rather than duplicate-impossible. Kafka allows manual control over when a record is considered consumed, but a crash after Temporal accepts the signal and before the offset is committed can still trigger redelivery. A practical way to make that safe is to keep the Workflow ID stable for the business entity and to make Activities idempotent, because Temporal may retry Activity executions as part of normal failure handling. Failure Semantics Matter More Than Labels The most common architectural mistake in Kafka and Temporal systems is to over-claim exactly-once semantics. Kafka’s idempotent producer ensures that retries do not create duplicate writes in the stream, and Kafka transactions allow atomic writes across partitions and topics. Kafka Streams goes further by defining end-to-end exactly-once around a very specific boundary: input topic offsets, state stores, and output topics are committed atomically because they are all inside Kafka’s storage model. Temporal, meanwhile, gives an effectively once-scheduled experience for Activities, but still expects Activity implementations to be idempotent because retries can occur after partial execution or worker failure. The combined system, therefore, does not become end-to-end exactly-once by default; that only happens when idempotency keys or transactional guarantees explicitly cover every external side effect that matters. Java public void publishConfirmed(String orderId) { producer.beginTransaction(); try { producer.send(new ProducerRecord<>("order-confirmed", orderId, orderId)).get(); producer.commitTransaction(); } catch (Exception ex) { producer.abortTransaction(); throw ex; } } This kind of publishing Activity is useful when Workflow progress must result in one or more Kafka records that either all appear or all fail together. The producer should be configured for idempotence, durable acknowledgments, and a transactional.id, but the design should still assume that non-Kafka side effects may need compensation. Temporal’s error-handling guidance recommends rollback logic with the Saga pattern for multi-step processes, which maps naturally to workflows that can reserve inventory, attempt payment, publish status, and then compensate in reverse order if one boundary fails after another has already succeeded. Long-Running Streams Need Long-Running Discipline Once Kafka is feeding entity-centric workflows for days or weeks, operational details start to matter as much as API design. Reusing the same business key as the Kafka record key and the Temporal Workflow ID creates a clean ownership model: Kafka uses keys to select partitions, partitions remain totally ordered, and Temporal guarantees that only one Workflow Execution with a given ID is open at a time. That alignment naturally serializes updates for a customer, order, or account across both systems. At the same time, the Kafka side of the bridge should stay thin enough to keep polling regularly, because consumers that stop polling can be considered dead and rebalanced out of the group. Temporal workflows that receive large numbers of Signals or perform many Activity calls also need history management. Event History is the mechanism that makes recovery possible, but it has performance limits and hard ceilings; Temporal warns as history grows and recommends Continue-As-New for long-running executions or workloads that process thousands of events. That becomes especially important in Kafka-driven entity workflows, where a single logical process can become a permanent mailbox unless it periodically rolls forward into a fresh run. Code evolution must also be handled deliberately because workflow logic is replayed; Temporal’s versioning guidance requires patching or worker versioning when changes would otherwise introduce non-determinism for in-flight executions. Conclusion Temporal and Kafka work best together when each is allowed to solve the problem it was built for. Kafka should distribute ordered, replayable events across the system boundary, and Temporal should hold the durable state machine that decides what those events mean over time. With that separation, retries stop leaking into application code, timers stop depending on process uptime, and compensations stop turning into chains of callbacks and ad hoc status flags. The result is not merely a system that survives failures, but a system whose failure semantics remain understandable under load, redelivery, redeployments, and long-running business latency.

By Akhil Madineni

Big Data

DZone's Featured Big Data Resources

Top Big Data Experts

The Latest Big Data Topics