Big Data

AWS Glue makes it easy to get a PySpark pipeline running quickly. It is significantly harder to build one that stays maintainable as logic grows, performs reliably at scale, and does not quietly accumulate operational debt over time. Most Glue pipelines start simple and become difficult to manage gradually — formulas get hardcoded, modules grow without boundaries, output files proliferate, and before long a single job is doing too many things in ways that are hard to test, hard to debug, and expensive to change. This article presents a set of design principles drawn from production Glue ETL pipelines processing billions of rows. Each principle is independent — you do not need to adopt all of them to benefit from any one. But together they form a coherent approach to building Glue pipelines that are modular, observable, cost-efficient, and built to last. Principle 1: Externalize Logic Into Config, Not Code The single most impactful structural decision in a Glue pipeline is where business logic lives. When formulas, dataset references, column selections, and filter conditions are hardcoded in PySpark, every change requires modifying job code, redeploying, and re-validating the full pipeline. A one-line formula change carries the same deployment risk as a structural refactor. Over time, this creates a strong disincentive to make changes, and the pipeline calcifies. The better pattern is to treat the Spark job as a generic executor and externalize all business-specific declarations into configuration. Formulas are declared as config entries with operands, rounding rules, and output names. Dataset loading behavior — which table, which columns, which filters, whether to cache — is declared per source rather than scripted per job. Schema shapes for complex types are declared explicitly rather than inlined. JSON { "source_table": "headcount_actuals", "database": "finance_db", "select_columns": ["site", "badge_type", "headcount", "fiscal_week"], "filters": [{"column": "is_active", "value": "Y"}], "rename": {"hc_count": "headcount"}, "cache": true } When a new dataset is needed, a new config entry is added — no Spark code changes. When a formula changes, the config entry is updated — no job redeployment required. The job itself becomes stable and generic; only config changes as business requirements evolve. This principle pays increasing dividends over time. Pipelines with externalized logic are faster to modify, safer to deploy, and easier to hand off because the business rules are readable independently of the execution engine. Principle 2: Design Modules With Explicit Boundaries A Glue job that does everything in one place is easy to write and hard to maintain. As pipelines grow, the instinct to add more logic to an existing job accelerates technical debt faster than almost any other decision. The more durable pattern is to decompose computation into modules with explicit input and output contracts. Each module receives one or more DataFrames, applies a focused set of transformations, and produces a named output DataFrame. Modules communicate exclusively through in-memory DataFrame references — there is no disk I/O between stages, no shared mutable state, and no implicit dependency on execution order beyond what the data flow itself requires. Utilities follow the same boundary principle, organized into two layers. Generic pipeline utilities handle cross-cutting concerns — file writing, dataset loading, filtering, deduplication, pivot operations — and are shared across all modules. Module-specific utilities implement transformation logic scoped to a single module and are never invoked outside it. This structure means adding a new module requires only writing its scoped utilities and wiring it into the pipeline. The generic layer is never touched. Existing modules are never at risk from new module development. The downstream benefit is testability. Each module with clean boundaries can be validated independently using mocked PySpark DataFrames with no Glue environment required. Engineers can run pytest locally against individual modules, iterate quickly, and deploy only after local validation passes. Principle 3: Choose Your Job Topology Deliberately A common default in complex pipelines is to split computation across multiple Glue jobs, using S3 as the handoff layer between stages. This is sometimes the right choice — but it should be a deliberate decision, not an instinct. Multi-job topologies make sense when stages have genuinely different compute profiles, when intermediate outputs need to be reused independently by other consumers, or when a stage failure should not force a full recompute from the beginning. In these cases, job separation gives you independent retry boundaries, independent DPU sizing, and the ability to schedule stages on different cadences. Single-job topologies — where the full pipeline runs within one Spark session — make sense when all computation is tightly coupled, modules share the same input datasets, and intermediate outputs have no standalone value. Running everything in one session eliminates cold start overhead for intermediate stages, avoids the cost of serializing data to S3 and deserializing it back between jobs, and keeps the execution model simple to reason about: one trigger, one job, one result. The question to ask is whether the stages truly need to be independent. If intermediate S3 persistence adds coordination complexity without adding value — no independent consumers, no differential retry requirements, no meaningful DPU difference between stages — then collapsing to a single job is usually faster, simpler, and cheaper. If stages have real independence requirements, splitting them is the right call and the operational overhead is justified. Neither topology is inherently superior. The mistake is defaulting to one without evaluating the trade-offs for the specific pipeline at hand. Principle 4: Overlap Writes With Computation When Latency Matters Overlapping writes with computation is a well-established technique in high-performance computing, deep learning training, and heavy database operations. The core idea is to hide the slow latency of I/O operations by running them in the background while the CPU or GPU continues processing data. Rather than waiting for a write to complete before starting the next computation, both proceed simultaneously — I/O latency is absorbed into computation time rather than added on top of it. In Glue ETL pipelines, the same principle applies directly. In a pipeline where multiple output DataFrames are produced, the naive write strategy — complete all computation, then write all outputs sequentially — has two compounding problems. First, it creates a peak memory spike: all computed results are held in memory simultaneously while writes proceed one by one. Second, it serializes work that does not need to be serial: every millisecond spent waiting for S3 acknowledgment is a millisecond the Spark executors are idle. This is worth addressing only when latency is a meaningful constraint. For low-frequency batch jobs running overnight with no user-facing SLA, sequential writes are perfectly adequate. But for pipelines where users or downstream systems are waiting on results — or where job duration directly affects infrastructure cost — overlapping writes with computation delivers measurable wall-clock reduction. The two-phase write strategy implements this directly. Outputs from early modules are written to S3 in background threads immediately after those modules complete, running in parallel with later computation stages. By the time all computation finishes, a significant portion of the output data has already landed in S3. Remaining outputs are then flushed concurrently in a second phase. The implementation leans on Python's concurrent.futures.ThreadPoolExecutor to manage background write threads while the main Spark session continues computation on the driver. A generic write orchestration utility can wrap this pattern so individual modules never need to manage thread lifecycle directly — they simply declare their output and the utility handles scheduling, thread management, and error propagation. Python from concurrent.futures import ThreadPoolExecutor, as_completed def write_phase_a(write_tasks): with ThreadPoolExecutor(max_workers=len(write_tasks)) as executor: futures = {executor.submit(task["fn"], task["df"], task["path"]): task["name"] for task in write_tasks} for future in as_completed(futures): name = futures[future] future.result() logger.info(f"[Phase A] Write complete: {name}") The practical effect is that peak memory pressure is distributed over the job's lifetime rather than concentrated at the end, and total wall-clock time is reduced by the overlap between I/O and CPU-bound computation. For pipelines with many output datasets and a latency SLA to meet, the savings compound significantly. Principle 5: Right-Size Output Files With a Reusable Writer Utility Right-sizing output files is the practice of tuning file sizes to balance disk I/O performance, network transfer speeds, and downstream processing efficiency. Too many small files and downstream readers spend more time on metadata operations and S3 API calls than on actual data reads. Too few large files and parallelism suffers — readers cannot split work efficiently across threads or nodes. The target is consolidated, evenly sized files that match the read patterns of downstream consumers. Spark's default output behavior writes one file per partition, and partition counts are typically tuned for computation throughput rather than output shape. A job optimized for shuffle performance might produce hundreds of partitions, each containing a few megabytes of output data — perfectly reasonable for Spark internals, but harmful for any reader that comes after. This small file problem compounds over time as output partitions accumulate in S3 and the Glue Catalog metadata grows with them. The fix is a reusable writer utility that decouples output file sizing from Spark's internal partition count. Rather than accepting the default, the utility estimates the DataFrame's actual size, calculates the appropriate number of output files for a target file size — typically 128MB to 256MB per file — and coalesces partitions before writing. Python def write_optimized(df, output_path, partition_cols, target_file_size_mb=128): estimated_size_mb = df.rdd.map(lambda row: len(str(row))).sum() / (1024 * 1024) optimal_partitions = max(1, int(estimated_size_mb / target_file_size_mb)) df.coalesce(optimal_partitions) \ .write \ .partitionBy(*partition_cols) \ .parquet(output_path, mode="overwrite") Making this a shared generic utility rather than inline logic in each module has two practical benefits. First, it enforces consistent file sizing behavior across all outputs in the pipeline — no module accidentally writes thousands of tiny files because an engineer forgot to coalesce. Second, it centralizes the tuning knob: when the target file size needs to change — because downstream query patterns shift or a new consumer has different read characteristics — it changes in one place and applies everywhere. Right-sized output files improve Athena scan performance, reduce per-query S3 API costs, keep Glue Catalog partition metadata manageable, and make the output data easier to consume for any downstream system reading from S3. This is a low-effort, high-payoff improvement that applies to virtually every Glue pipeline writing to S3. Principle 6: Use Complex Types to Defer Denormalization SQL-based pipelines are constrained to flat, fully denormalized row structures at every intermediate stage because SQL has no native complex type support. This forces denormalization to happen early, inflating data volume at every subsequent join and aggregation. PySpark has native support for structs, maps, and arrays. Using these types at intermediate stages allows related values to be grouped logically without inflating row counts. A row that would require five denormalized rows in SQL can be represented as a single row with a struct or array column in Spark. Denormalization is then deferred to the final output layer only — applied once, at write time, for consumers that require flat structures. Everything upstream of the final write benefits from reduced volume, fewer shuffles, and faster joins. This principle is particularly impactful in pipelines with multi-level aggregations or wide schemas where dozens of metrics attach to the same dimensional key. Keeping those metrics grouped in a struct until the final output stage reduces the effective row count and join complexity throughout the pipeline. Principle 7: Build Observability Into Every Stage Glue jobs that fail silently or surface errors as opaque stack traces at the end of a long execution are expensive to debug. The investment in step-level observability pays back quickly the first time something goes wrong in production. The minimum viable observability pattern is row count logging at every materialization point. After each module completes and after each write, log the output row count with a descriptive label. This gives a running picture of data volume through the pipeline and makes it immediately obvious when a transformation has dropped rows unexpectedly or produced more rows than expected. Python def log_step(df, step_name): count = df.count() logger.info(f"[{step_name}] Row count: {count:,}") return df Pair this with a try/except/finally pattern at the job level that ensures spark.catalog.clearCache() is always called on exit — whether the job succeeds or fails — to release cached DataFrames and avoid memory leaks across retries. Python try: run_pipeline() except Exception as e: logger.error(f"Pipeline failed: {e}") raise finally: spark.catalog.clearCache() CloudWatch captures all logs automatically. When a job fails, the row count trail shows exactly where in the pipeline the problem occurred, making triage faster and reducing the time between failure and fix. Principle 8: Isolate Executions for Concurrency Pipelines that share compute resources across simultaneous executions create contention that is difficult to predict and expensive to manage. The common response — queue-based serialization — adds operational complexity without solving the underlying resource constraint. AWS Glue's execution model eliminates this problem structurally. Each job execution gets its own isolated DPU allocation. There is no shared compute pool. Ten simultaneous executions consume ten independent DPU allocations and do not interfere with each other in any way. Designing for this means treating each execution as fully independent: no shared state, no cross-execution coordination, no assumption about what other executions are running. Combined with idempotent writes — using overwrite mode so a retry produces the same result as the original execution — the pipeline becomes safe to run concurrently at any scale without additional coordination logic. The cost model reinforces this. Glue bills per DPU-second of actual compute consumed. An execution that takes eight minutes on 240 DPUs costs the same whether it runs alone or alongside a hundred other executions. There is no premium for concurrency and no shared pool to provision for peak load. Putting It Together These eight principles are independent but complementary. A pipeline that applies all of them is modular enough to develop in parallel, observable enough to debug quickly, cost-efficient enough to run at scale, and stable enough to maintain over time without accumulating structural debt. The quickest wins for most existing pipelines are Principles 1, 5, and 7 — externalizing logic into config, right-sizing output files with a shared utility, and adding row count logging at every stage. Each can be applied incrementally without restructuring the full pipeline. The remaining principles become more valuable as pipeline complexity grows and concurrency requirements increase. The underlying thesis is simple: a well-designed Glue pipeline should be easy to change, easy to test, easy to debug, and cheap to run. None of those properties require exotic infrastructure. They require deliberate design decisions applied consistently from the start.

By Janani Annur Thiruvengadam 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

By Jubin Soni, FBCS CORE

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.

Raw data doesn't win model competitions. Features do. And when your raw data is tens of billions of rows sitting across multiple sources, you can't afford to run pandas in a notebook and call it a day. In this tutorial, I'll walk through building a production-grade feature engineering pipeline on Azure Databricks using: Apache Spark for distributed transformation at scaleDelta Lake for reliable, versioned feature storage with ACID guaranteesMLflow for tracking feature pipeline runs, parameters, and the models trained on top of them The use case is a customer churn prediction system, but the patterns apply to any ML feature pipeline. Architecture Overview The pipeline follows the Medallion Architecture — a layered approach where data gets progressively cleaner and more feature-ready as it moves from Bronze to Silver to Gold. MLflow sits across all three layers, tracking every run. Pipeline Flow Layer Breakdown LayerDelta TableWhat happens hereTypical latencyBronzechurn.bronze.eventsRaw ingest, no transforms, append onlyMinutesSilverchurn.silver.customersDeduplication, null handling, schema enforcementMinutesGoldchurn.gold.featuresAggregations, window functions, encodingMinutes to hoursMLflow RunN/ATraining, metric logging, artifact storageHoursRegistryN/AVersioned model store, stage promotionOn demand Step 1 — Bronze Layer: Raw Ingest The Bronze layer is append-only. No transforms. No business logic. Just get the data in and preserve it exactly as it arrived so you can always replay from source. Python from pyspark.sql import SparkSession from pyspark.sql.functions import current_timestamp, lit from delta.tables import DeltaTable spark = SparkSession.builder.getOrCreate() # Read raw events from ADLS Gen2 / Event Hub / source of choice raw_events = spark.read.format('json').load('abfss://[email protected]/events/') # Add ingestion metadata — never mutate source columns bronze_df = raw_events.withColumn('_ingested_at', current_timestamp()) \ .withColumn('_source', lit('events_api')) # Write to Bronze Delta table — append only, no overwrites bronze_df.write \ .format('delta') \ .mode('append') \ .option('mergeSchema', 'true') \ .saveAsTable('churn.bronze.events') print(f"Bronze rows written: {bronze_df.count()}") Why append-only? If your downstream pipeline produces bad features, you want to replay from Bronze without re-ingesting from source. Overwriting Bronze breaks that ability. Step 2 — Silver Layer: Clean and Validate Silver is where you enforce schema, handle nulls, deduplicate, and standardize. Think of it as your canonical, trusted dataset. Python from pyspark.sql.functions import col, to_timestamp, when, trim, upper from delta.tables import DeltaTable bronze = spark.table('churn.bronze.events') silver_df = bronze \ .filter(col('customer_id').isNotNull()) \ .filter(col('event_type').isNotNull()) \ .dropDuplicates(['customer_id', 'event_id']) \ .withColumn('event_ts', to_timestamp(col('event_timestamp'))) \ .withColumn('event_type', upper(trim(col('event_type')))) \ .withColumn('country_code', when(col('country').isNull(), lit('UNKNOWN')) .otherwise(upper(col('country')))) \ .select( 'customer_id', 'event_id', 'event_type', 'event_ts', 'country_code', 'product_id', 'session_id', '_ingested_at', ) # Upsert into Silver using Delta MERGE — idempotent on re-runs if DeltaTable.isDeltaTable(spark, 'churn.silver.customers'): silver_table = DeltaTable.forName(spark, 'churn.silver.customers') silver_table.alias('tgt').merge( silver_df.alias('src'), 'tgt.customer_id = src.customer_id AND tgt.event_id = src.event_id' ).whenNotMatchedInsertAll().execute() else: silver_df.write.format('delta').saveAsTable('churn.silver.customers') print(f"Silver table updated. Total rows: {spark.table('churn.silver.customers').count()}") Step 3 — Gold Layer: Feature Engineering This is the heart of the pipeline. We compute aggregated, windowed, and encoded features that the model will actually train on. Python from pyspark.sql.functions import ( col, count, countDistinct, sum as _sum, avg, datediff, max as _max, min as _min, current_date, expr, when ) from pyspark.sql.window import Window silver = spark.table('churn.silver.customers') # ------------------------------------------------------------------ # 1. Aggregate features per customer over 30 / 90 day windows # ------------------------------------------------------------------ today = current_date() agg_features = silver \ .withColumn('days_since_event', datediff(today, col('event_ts'))) \ .groupBy('customer_id') \ .agg( count('event_id') .alias('total_events'), countDistinct('session_id') .alias('total_sessions'), countDistinct('product_id') .alias('distinct_products'), _sum(when(col('days_since_event') <= 30, 1).otherwise(0)) .alias('events_last_30d'), _sum(when(col('days_since_event') <= 90, 1).otherwise(0)) .alias('events_last_90d'), _max('event_ts') .alias('last_event_ts'), _min('event_ts') .alias('first_event_ts'), ) \ .withColumn('days_since_last_event', datediff(today, col('last_event_ts'))) \ .withColumn('customer_tenure_days', datediff(today, col('first_event_ts'))) \ .withColumn('avg_events_per_day', col('total_events') / (col('customer_tenure_days') + 1)) # ------------------------------------------------------------------ # 2. Encode churn risk tier as ordinal feature # ------------------------------------------------------------------ feature_df = agg_features \ .withColumn('recency_tier', when(col('days_since_last_event') <= 7, lit(3)) # active .when(col('days_since_last_event') <= 30, lit(2)) # at risk .otherwise(lit(1)) # churned ) \ .withColumn('engagement_score', (col('events_last_30d') * 0.6 + col('events_last_90d') * 0.4) / (col('customer_tenure_days') + 1) ) # ------------------------------------------------------------------ # 3. Write to Gold feature store — overwrite with partition by date # ------------------------------------------------------------------ feature_df \ .withColumn('feature_date', current_date()) \ .write \ .format('delta') \ .mode('overwrite') \ .option('replaceWhere', f"feature_date = '{today}'") \ .saveAsTable('churn.gold.features') print(f"Gold features written: {feature_df.count()} customers") Step 4 — MLflow: Track the Training Run With features in Gold, we hand off to MLflow to train, track, and register the model. Notice we log the Delta table version so we can always reproduce exactly which feature snapshot trained which model. Python import mlflow import mlflow.sklearn from mlflow.models.signature import infer_signature from sklearn.ensemble import GradientBoostingClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import roc_auc_score, f1_score import pandas as pd mlflow.set_experiment('/churn-prediction/feature-pipeline') # Read Gold features — capture Delta version for reproducibility gold_table = DeltaTable.forName(spark, 'churn.gold.features') delta_version = gold_table.history(1).select('version').collect()[0][0] features_pdf = spark.table('churn.gold.features').toPandas() FEATURE_COLS = [ 'total_events', 'total_sessions', 'distinct_products', 'events_last_30d', 'events_last_90d', 'days_since_last_event', 'customer_tenure_days', 'avg_events_per_day', 'recency_tier', 'engagement_score', ] TARGET = 'churned' X = features_pdf[FEATURE_COLS] y = features_pdf[TARGET] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) with mlflow.start_run(run_name=f'gbm-features-v{delta_version}') as run: params = {'n_estimators': 200, 'max_depth': 5, 'learning_rate': 0.05} model = GradientBoostingClassifier(**params, random_state=42) model.fit(X_train, y_train) y_pred = model.predict(X_test) y_prob = model.predict_proba(X_test)[:, 1] # Log everything mlflow.log_params(params) mlflow.log_metric('roc_auc', roc_auc_score(y_test, y_prob)) mlflow.log_metric('f1_score', f1_score(y_test, y_pred)) mlflow.log_param('delta_feature_version', delta_version) mlflow.log_param('feature_columns', FEATURE_COLS) mlflow.log_param('training_rows', len(X_train)) # Log model with signature signature = infer_signature(X_train, y_pred) mlflow.sklearn.log_model( model, artifact_path='churn-gbm', signature=signature, registered_model_name='churn-prediction-gbm', ) print(f"Run ID: {run.info.run_id}") print(f"ROC-AUC: {roc_auc_score(y_test, y_prob):.4f}") print(f"Feature Delta version logged: {delta_version}") Bonus: Delta Lake Time Travel for Feature Reproducibility One of the best things about Delta Lake is time travel. If a model behaves unexpectedly in production, you can reload the exact feature snapshot it was trained on. Python # Reload the exact feature version that trained a specific model run import mlflow run = mlflow.get_run('your-run-id-here') feature_version = int(run.data.params['delta_feature_version']) # Rehydrate that exact feature snapshot historical_features = spark.read \ .format('delta') \ .option('versionAsOf', feature_version) \ .table('churn.gold.features') print(f"Loaded feature snapshot from Delta version {feature_version}") print(f"Row count: {historical_features.count()}") # You can now retrain on the exact same data to reproduce the result Service Comparison ToolRole in pipelineWhy not the alternativeApache SparkDistributed feature computationPandas (single node, OOM at scale), Dask (less native Databricks integration)Delta LakeFeature storage with versioningParquet (no ACID, no time travel), Hive tables (no merge support)MLflow TrackingExperiment and param loggingManual logging (not reproducible), W&B (extra cost, less native on Databricks)MLflow RegistryModel versioning and promotionCustom model store (more ops overhead)Medallion ArchitecturePipeline layer separationFlat pipelines (hard to debug, no replay capability)Delta MERGEIdempotent Silver upsertsOverwrite (destroys history), append (creates duplicates) Things to Watch in Production Shuffle partitions matter. Spark defaults to 200 shuffle partitions, which is fine for small data but will bottleneck at scale. Set spark.conf.set("spark.sql.shuffle.partitions", "auto") on Databricks Runtime 10+ or tune it manually to 2-3x your core count. Z-ordering on Gold features. If you're querying Gold by customer_id frequently, add OPTIMIZE churn.gold.features ZORDER BY (customer_id) after the write. This co-locates related data and cuts query times dramatically on large tables. Log Delta version in every MLflow run. This is non-negotiable for reproducibility. Without it you can't prove which feature snapshot trained which model, which becomes a compliance problem in regulated industries. Cluster autoscaling for feature jobs. Feature engineering jobs tend to have spiky resource needs (big during aggregation, small during writes). Enable autoscaling on your Databricks cluster and set a min/max node count rather than a fixed size. Wrapping Up The combination of Spark, Delta Lake, and MLflow on Databricks gives you a feature engineering pipeline that is reproducible (Delta time travel + MLflow param logging), scalable (Spark handles billions of rows), and auditable (every run is tracked, every feature version is stored). The Medallion Architecture keeps the pipeline modular — you can rerun just the Gold layer if you change a feature definition without touching Bronze or Silver, and MLflow ties model performance back to the exact feature version that produced it. References Azure Databricks DocumentationDelta Lake — The Definitive GuideApache Spark SQL — Window FunctionsMLflow Tracking DocumentationMLflow Model RegistryMedallion Architecture on DatabricksDelta Lake Time TravelDatabricks Feature Store Overview

By Jubin Soni, FBCS CORE

Industrial control systems are generating more data than ever before, but the Python tooling used to process this telemetry often encounters severe performance constraints. Traditional OPC UA libraries are built around synchronous, polling-based Client and Server architectures. When industrial networks scale to thousands of sensors broadcasting high-frequency data, these synchronous Python implementations choke. To handle this modern many-to-many topology, developers need a native Publisher and Subscriber solution that does not block the execution thread while waiting for network packets. For Python developers unfamiliar with industrial protocols, OPC UA PubSub (IEC 62541-14) is a standard that decouples data producers from consumers by allowing devices to broadcast telemetry via stateless middleware like UDP Multicast. For industrial engineers new to Python concurrency, asyncio is a standard library that uses an event loop to handle thousands of simultaneous network operations concurrently without the heavy overhead of traditional threading. Bridging these two paradigms requires a completely non-blocking architecture. To address this gap, a complete asyncio driven OPC UA PubSub implementation was architected and integrated into the open source opcua-asyncio library (merged in Commit 2b6f3e5). Implementing this standard from scratch in an asynchronous Python environment presented unique challenges. This article breaks down the engineering decisions and technical design patterns used to build this extension. By contributing this capability to a library that serves thousands of developers in the Python IIoT ecosystem, the goal is to ensure engineers can now build highly scalable publisher and subscriber sensor networks without migrating away from Python. The Shift to Publisher and Subscriber in IIoT In traditional OPC UA, a client polls a server or sets up monitored items. This creates a tightly coupled, connection-oriented topology. The PubSub extension decouples this by allowing publishers to broadcast telemetry data via stateless middleware like UDP Multicast or MQTT, which subscribers can passively ingest. To bring this to the opcua-asyncio ecosystem, the architecture needed to bridge the gap between Python's asynchronous event loop and the highly deterministic, byte-packed UADP (OPC UA Datagram Protocol) structures. The design was broken down into four core pillars. Asynchronous transport layer: Managing non-blocking UDP and IP multicast.UADP binary protocol engine: Bit-level packing and unpacking of network messages.Data abstraction and node mapping: Linking arbitrary network payloads to the OPC UA Address Space.Concurrency and connection management: Orchestrating readers, writers, and tasks via asyncio. Pillar 1: The Asynchronous UDP Transport Layer OPC UA UADP relies on UDP for low-latency transmission. In Python, synchronous socket operations block the main thread, which is fatal to an asyncio application. To solve this, the networking layer was built directly on top of asyncio.DatagramProtocol. The OpcUdp class overrides the standard protocol callbacks to bridge the network socket with the PubSub receiver logic. Here is a look at how the protocol was extended and hooked into the event loop to ensure incoming datagrams never block the main thread. Python class OpcUdp(asyncio.DatagramProtocol): def __init__(self, cfg: UdpSettings, receiver: Optional[PubSubReceiver], publisher_id: Variant) -> None: super().__init__() self.cfg = cfg self.receiver = receiver self.publisher_id = publisher_id.Value def datagram_received(self, data: bytes, source: Tuple[str, int]) -> None: try: buffer = Buffer(data) msg = UadpNetworkMessage.from_binary(buffer) if self.receiver is not None: asyncio.ensure_future(self.receiver.got_uadp(msg)) except Exception: logging.exception("Received Invalid UadpPacket") Socket lifecycle: The UdpSettings class manages socket creation by carefully applying SO_REUSEADDR and handling both IPv4 (AF_INET) and IPv6 (AF_INET6) multicasting.Multicast configuration: Depending on the IP family, IP_ADD_MEMBERSHIP or IPV6_JOIN_GROUP are injected directly into the socket options via the struct module to ensure the application correctly subscribes to IGMP or MLD groups.Non-blocking reception: When a datagram hits the interface, datagram_received immediately passes the raw bytes to the UADP decoding engine and dispatches the resulting parsed message to a background task using asyncio.ensure_future(). This guarantees the networking thread is instantly freed to handle the next packet. Pillar 2: The UADP Binary Protocol Engine The UADP specification defines an extremely dense, highly variable network packet. Headers can dynamically expand or contract based on a series of bit flags. Processing this in Python requires rigorous byte manipulation to maintain both memory efficiency and processing speed. The uadp.py implementation utilizes Python's enum.IntFlag to map the exact bitwise schemas defined in OPC UA Part 14. Python class MessageHeaderFlags(IntFlag): NONE = 0 UADP_VERSION_BIT0 = 0b1 PUBLISHER_ID = 0b00010000 GROUP_HEADER = 0b00100000 PAYLOAD_HEADER = 0b01000000 EXTENDED_FLAGS_1 = 0b10000000 # FlagsExtend1 PUBLISHER_ID_UINT16 = 0b0000000100000000 PUBLISHER_ID_UINT32 = 0b0000001000000000 PUBLISHER_ID_UINT64 = 0b0000011000000000 PUBLISHER_ID_STRING = 0b0000010000000000 Flag-driven serialization: The UadpHeader and UadpDataSetMessageHeader are deeply nested and conditional. For example, the Extended Flags dictate whether a PublisherId is encoded as a Byte, UInt16, UInt32, UInt64, or String.Bitwise extensibility: The implementation cascades flags using EXTENDED_FLAGS_1 and EXTENDED_FLAGS_2 bits. If the integer value of the required flags exceeds 0xFF, the engine dynamically shifts the bytes and appends the extension flags.Binary packing: A standardized Primitives unpacking utility translates the raw buffer directly into strictly typed Python objects like UInt32, Guid, or DateTime. This avoids the overhead of intermediate object instantiation when parsing high-frequency sensor data.Delta Frames vs. raw data: The parser dynamically routes payload deserialization based on MessageDataSetFlags. It distinguishes between Key Frames, Delta Frames, and Raw Data while packing the resulting generic DataValue structs into a unified UadpNetworkMessage. Pillar 3: Data Abstraction and Address Space Integration Receiving data is only half the battle because that data must meaningfully map to the server's Address Space. The architecture introduces PubSubInformationModel to handle this synchronization. Datasets and metadata: A PublishedDataSet defines the structure of the data being transmitted. This includes tracking FieldMetaData, built in types, and value ranks.Dynamic variable substitution: The PubSubDataSourceServer class abstracts the retrieval of data from the server tree. It safely reads attributes and falls back to a SubstituteValue if a node status code is bad. This ensures unbroken telemetric streams.Subscribed mirrors: When an OPC UA client acts as a subscriber, it needs to see the incoming data reflected in its own node tree. The SubscribedDataSetMirror dynamically creates new variable nodes on the fly to match the incoming DataSetMetaData. This dynamic node mapping was engineered by injecting new variables straight into the server tree based on the metadata specification. Python async def _create_and_set_node(self, f: FieldMetaData): if self._node is None: raise RuntimeError("SubscribedDataSetMirror._node is not initialized.") n = await self._node.add_variable( NodeId(NamespaceIndex=Int16(1)), "1:" + str(f.Name), Variant(), datatype=f.DataType ) await n.write_attribute(AttributeIds.Description, f.Description) await n.write_attribute(AttributeIds.ValueRank, f.ValueRank) await n.write_attribute(AttributeIds.ArrayDimensions, f.ArrayDimensions) return n Target variables: Alternatively, SubScribedTargetVariables maps incoming dataset fields directly to existing NodeId references in the server. These references update in real time as UDP packets are decoded. Pillar 4: Concurrency and Connection Management The top-level orchestration is handled by the PubSubConnection and PubSub classes. These act as the asynchronous lifecycle managers. Task gathering: When start() is invoked on a connection, the lifecycle manager utilizes asyncio.gather() to concurrently spin up all associated DataSetReader and DataSetWriter tasks without blocking the main OPC UA server loop. Python async def start(self) -> None: logging.info("Starting Connection %s", await self.get_name()) loop = asyncio.get_event_loop() sock, _, _ = self._network_settings.create_socket() self._transport, self._protocol = await loop.create_datagram_endpoint( lambda: self._network_factory(self._network_settings, self._receiver, self._cfg.PublisherId), sock=sock, ) self._writer_tasks = asyncio.gather(*[writer.run(self._protocol, self._app) for writer in self._writer_groups]) reader_tasks = asyncio.gather(*[reader.start() for reader in self._reader_groups]) await reader_tasks if self._protocol is not None: self._protocol.set_receiver(self._receiver) await self._set_state(PubSubState.Operational) Protocol decoupling: To prevent circular dependencies between the network transport and the information model, strict interfaces defined in protocols.py are used. The UDP protocol layer communicates with the logical layer strictly through these abstract protocols.Wildcard routing and readers: The ReaderGroup acts as an intelligent multiplexer. When a multi-payload UADP packet arrives, it analyzes the GroupHeader and DataSetPayloadHeader. It then routes individual DataSetMessages to the correct DataSetReader instances by matching wildcard filters.Timeouts and state machines: Robust industrial systems must handle connection drops. The DataSetReader wraps its operation in a dedicated timeout task. Using asyncio.wait_for(), it actively monitors for MessageReceiveTimeout events. If a heartbeat or payload is missed, it transitions the internal PubSubState to Error. This allows higher-level application logic to gracefully degrade. Conclusion Building a production-ready OPC UA PubSub stack in Python requires harmonizing the stringent bit-packed demands of the IEC 62541-14 specification with the asynchronous paradigms of asyncio. By leveraging asyncio.DatagramProtocol for deterministic networking, abstracting the UADP bit flags into structured classes, and deeply integrating with the OPC UA Address space via mirrored target variables, this implementation provides a scalable foundation for modern IIoT architectures. Code and Open Source Contributions The architecture and implementation details discussed in this article were merged into the core FreeOpcUa/opcua-asyncio repository. You can explore the complete implementation, including the raw protocol parsing and asyncio abstractions, via the links below. Primary commit: 2b6f3e5 (Initial implementation of OPC UA PubSub UDP and UADP). Key files to explore in the commit: asyncua/pubsub/udp.py: Contains the OpcUdp transport layer and multicast socket configuration.asyncua/pubsub/uadp.py: Houses the flag driven serialization and binary protocol engine.asyncua/pubsub/connection.py: Demonstrates the asyncio task management and lifecycle orchestration.

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.

The Feature Engineering Problem Feature engineering is where most ML projects silently fail in production. Not because the model is wrong — but because the features the model sees at training time are different from the features it sees at inference time. This is called training-serving skew, and it's the #1 silent killer of ML systems. Three specific failure modes cause it: Online/offline inconsistency – the batch pipeline that computes training features uses different logic than the real-time service that computes inference featuresData leakage – training features accidentally include information from the future (e.g., joining on a label that was created after the event)Feature staleness – a model trained on 30-day rolling averages is served features that are 6 hours stale because the pipeline backfills are slow The Databricks Feature Store — now part of Unity Catalog as Feature Engineering in Unity Catalog — solves all three by: Storing feature computation logic alongside the data (no drift between training and serving)Enforcing point-in-time lookups during training dataset creationProviding a unified API for both batch offline reads and low-latency online reads Architecture Overview Feature Store Concepts: ERD Understanding the data model behind the Feature Store is essential for designing correct pipelines. Here's how the entities relate: The critical relationship: a Model Version is bound to a Training Set, which records exactly which feature tables and which point-in-time lookups were used. This is how Databricks guarantees reproducibility — you can always re-create the exact training data that produced any model version. Environment Setup Python # Databricks Runtime ML 13.x+ recommended # Feature Engineering in Unity Catalog (formerly Feature Store) %pip install databricks-feature-engineering==0.6.0 --quiet dbutils.library.restartPython() from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup from databricks.feature_engineering.entities.feature_serving_endpoint import ( ServedEntity, EndpointCoreConfig ) from pyspark.sql import functions as F, SparkSession from pyspark.sql.types import ( StructType, StructField, StringType, LongType, DoubleType, TimestampType, ArrayType ) import mlflow spark = SparkSession.builder.getOrCreate() fe = FeatureEngineeringClient() # Unity Catalog paths CATALOG = "prod" FEATURE_DB = f"{CATALOG}.feature_store" EVENTS_TABLE = f"{CATALOG}.silver.events_clean" KAFKA_BROKER = "kafka-broker.internal:9092" KAFKA_TOPIC = "user-events" # Checkpoint locations (ADLS / S3 / GCS) CHECKPOINT_BASE = "abfss://[email protected]/features" Streaming Feature Pipeline The streaming pipeline reads from Kafka, computes windowed aggregations using Spark's stateful streaming engine, and writes features to the Feature Store via foreachBatch. This keeps the feature table continuously fresh. Python # ── Streaming Feature Pipeline ──────────────────────────────────────────────── # Step 1: Define the raw event schema from Kafka event_schema = StructType([ StructField("user_id", StringType(), False), StructField("event_type", StringType(), True), StructField("product_id", StringType(), True), StructField("revenue", DoubleType(), True), StructField("session_id", StringType(), True), StructField("platform", StringType(), True), StructField("event_ts", TimestampType(), False), ]) # Step 2: Read from Kafka raw_stream = ( spark.readStream .format("kafka") .option("kafka.bootstrap.servers", KAFKA_BROKER) .option("subscribe", KAFKA_TOPIC) .option("startingOffsets", "latest") .option("failOnDataLoss", "false") .load() .select( F.from_json(F.col("value").cast("string"), event_schema).alias("data"), F.col("timestamp").alias("kafka_ts") ) .select("data.*", "kafka_ts") ) # Step 3: Apply watermark and compute windowed features # Watermark: tolerate up to 10 minutes of late data windowed_features = ( raw_stream .withWatermark("event_ts", "10 minutes") .groupBy( F.col("user_id"), F.window(F.col("event_ts"), "1 hour", "15 minutes").alias("window") ) .agg( F.count("*").alias("event_count_1h"), F.sum(F.when(F.col("event_type") == "purchase", F.col("revenue")) .otherwise(0)).alias("revenue_1h"), F.countDistinct("session_id").alias("session_count_1h"), F.countDistinct("product_id").alias("unique_products_1h"), F.sum(F.when(F.col("event_type") == "purchase", 1) .otherwise(0)).alias("purchase_count_1h"), F.first("platform").alias("last_platform"), ) # Flatten window struct to scalar columns .withColumn("window_start", F.col("window.start")) .withColumn("window_end", F.col("window.end")) .withColumn("feature_ts", F.col("window.end")) # timestamp key for PIT lookup .drop("window") # Derived features .withColumn("conversion_rate_1h", F.when(F.col("event_count_1h") > 0, F.col("purchase_count_1h") / F.col("event_count_1h")) .otherwise(0.0)) .withColumn("avg_revenue_per_purchase_1h", F.when(F.col("purchase_count_1h") > 0, F.col("revenue_1h") / F.col("purchase_count_1h")) .otherwise(0.0)) ) # Step 4: Write to Feature Store via foreachBatch # foreachBatch gives us transactional writes per micro-batch def write_to_feature_store(batch_df, batch_id): """ Called on each micro-batch. Merges feature data into the Feature Store table using merge_on keys (user_id + feature_ts). """ if batch_df.isEmpty(): return fe.write_table( name=f"{FEATURE_DB}.user_activity_features", df=batch_df, mode="merge", # upsert: update existing, insert new ) print(f"Batch {batch_id}: wrote {batch_df.count()} feature rows") # Step 5: Create the feature table (idempotent — safe to re-run) try: fe.create_table( name=f"{FEATURE_DB}.user_activity_features", primary_keys=["user_id"], timestamp_keys=["feature_ts"], schema=windowed_features.schema, description=( "Real-time user activity features computed from event stream. " "1-hour sliding window, refreshed every 15 minutes. " "Primary key: user_id. Timestamp key: feature_ts (window end)." ), ) print("Feature table created.") except Exception: print("Feature table already exists — continuing.") # Step 6: Launch the streaming query streaming_query = ( windowed_features.writeStream .outputMode("update") # update mode for stateful aggregations .option("checkpointLocation", f"{CHECKPOINT_BASE}/user_activity") .trigger(processingTime="5 minutes") # micro-batch every 5 min .foreachBatch(write_to_feature_store) .start() ) print(f"Streaming query '{streaming_query.name}' running...") print(f"Status: {streaming_query.status}") Point-in-Time Correct Training Dataset Generation This is the most critical part of the Feature Store. When creating training data, we must join labels to features at the timestamp of the label event — not the current time. This prevents data leakage. Python # ── Point-in-Time Correct Training Dataset ──────────────────────────────────── # Step 1: Load the label dataset # Each row = one prediction target event, with the exact timestamp # at which a model would have needed to make a prediction. labels_df = ( spark.table(f"{CATALOG}.gold.churn_labels") .select( "user_id", "churn_label", # 0 = retained, 1 = churned F.col("observation_ts").alias("event_timestamp"), # point-in-time anchor "experiment_split" # train/val/test ) .filter(F.col("observation_ts") >= "2024-01-01") ) print(f"Label rows: {labels_df.count():,}") labels_df.show(5) # +----------+-----------+---------------------+-----------------+ # | user_id |churn_label| event_timestamp | experiment_split| # +----------+-----------+---------------------+-----------------+ # | u_123456 | 0 | 2024-03-15 14:22:00 | train | # | u_789012 | 1 | 2024-03-15 18:45:00 | train | # Step 2: Define feature lookups # as_of_timestamp=None → use the label's event_timestamp (point-in-time) # Databricks will join each label row to the feature values # that were valid at event_timestamp — not the latest values. feature_lookups = [ # User activity features — 1h window features from the streaming pipeline FeatureLookup( table_name=f"{FEATURE_DB}.user_activity_features", feature_names=[ "event_count_1h", "revenue_1h", "session_count_1h", "unique_products_1h", "purchase_count_1h", "conversion_rate_1h", "avg_revenue_per_purchase_1h", "last_platform", ], lookup_key="user_id", timestamp_lookup_key="event_timestamp", # ← PIT anchor ), # User profile features — slower-changing, from batch pipeline FeatureLookup( table_name=f"{FEATURE_DB}.user_profile_features", feature_names=[ "account_age_days", "lifetime_revenue", "preferred_category", "subscription_tier", ], lookup_key="user_id", timestamp_lookup_key="event_timestamp", # ← PIT anchor ), # Transaction aggregates — 30d and 90d rolling windows FeatureLookup( table_name=f"{FEATURE_DB}.transaction_features", feature_names=[ "purchase_count_30d", "purchase_count_90d", "avg_order_value_30d", "days_since_last_purchase", "category_diversity_score", ], lookup_key="user_id", timestamp_lookup_key="event_timestamp", ), ] # Step 3: Create training dataset (Feature Store handles the PIT join) training_set = fe.create_training_set( df=labels_df, feature_lookups=feature_lookups, label="churn_label", exclude_columns=["observation_ts", "experiment_split"], ) # The returned DataFrame has features + labels, PIT-correct training_df = training_set.load_df() print(f"Training rows: {training_df.count():,}") print(f"Training cols: {len(training_df.columns)}") training_df.show(3) # Step 4: Train model and log via Feature Store (preserves lineage!) from sklearn.ensemble import GradientBoostingClassifier import pandas as pd train_pdf = ( training_df .filter(F.col("experiment_split") == "train") .drop("experiment_split", "user_id") .fillna(0) .toPandas() ) X_train = train_pdf.drop(columns=["churn_label"]) y_train = train_pdf["churn_label"] model = GradientBoostingClassifier( n_estimators=300, learning_rate=0.05, max_depth=5, subsample=0.8, random_state=42, ) with mlflow.start_run(run_name="churn-gbm-v1") as run: model.fit(X_train, y_train) # Log model via Feature Store — this records the feature lineage fe.log_model( model=model, artifact_path="churn_model", flavor=mlflow.sklearn, training_set=training_set, # ← binds model to its feature lookups registered_model_name=f"{CATALOG}.ml.user_churn_model", ) print(f"Logged model with feature lineage. Run: {run.info.run_id}") Writing Features to the Online Store For real-time inference, the model needs features in milliseconds — not the seconds it takes to query Delta Lake. Databricks Feature Store can publish features to an online store (DynamoDB, Cosmos DB, MySQL, etc.) for low-latency reads. Python # ── Publish Features to Online Store ───────────────────────────────────────── # Online stores are configured per feature table. # Here we publish user_activity_features to DynamoDB for <5ms lookups. from databricks.feature_engineering.entities.feature_store_online_table import ( OnlineTable, OnlineTableSpec, TriggeredSchedulingPolicy ) # Create an online table spec (backed by a serverless real-time compute layer) online_table_spec = OnlineTableSpec( primary_key_columns=["user_id"], source_table_full_name=f"{FEATURE_DB}.user_activity_features", run_triggered=OnlineTableSpec.TriggeredSchedulingPolicy(), # sync on-demand # OR for continuous sync: # run_continuous=OnlineTableSpec.ContinuousSchedulingPolicy() ) # Create the online table (idempotent) online_table = fe.create_online_table(spec=online_table_spec) print(f"Online table: {online_table.name}") print(f"Status: {online_table.status.detailed_state}") # Trigger an initial sync from the offline Delta table to the online store fe.refresh_online_table(name=f"{FEATURE_DB}.user_activity_features") Serving Features at Inference Time At inference time, the Feature Store SDK performs automatic feature lookups, joining the incoming request data with features from the online store before passing them to the model. Python # ── Real-Time Feature Serving at Inference ──────────────────────────────────── import requests, json WORKSPACE_URL = "https://<workspace>.azuredatabricks.net" TOKEN = dbutils.secrets.get("prod-scope", "databricks-token") # Option 1: Model Serving with automatic feature lookup # When you logged the model with fe.log_model(), Databricks knows which # features to fetch. You only send the lookup key (user_id) at inference time. def predict_churn(user_ids: list) -> list: """ Send only user_id — the serving endpoint fetches features automatically from the online store and runs inference. """ payload = { "dataframe_records": [ {"user_id": uid} for uid in user_ids ] } resp = requests.post( f"{WORKSPACE_URL}/serving-endpoints/churn-predictor/invocations", headers={ "Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json", }, data=json.dumps(payload), timeout=5, ) resp.raise_for_status() return resp.json()["predictions"] # Example usage predictions = predict_churn(["u_123456", "u_789012", "u_345678"]) for uid, pred in zip(["u_123456", "u_789012", "u_345678"], predictions): print(f"{uid}: churn_probability = {pred:.4f}") # u_123456: churn_probability = 0.0821 # u_789012: churn_probability = 0.7643 # u_345678: churn_probability = 0.1209 # Option 2: Direct feature lookup via the Feature Serving endpoint # Useful when you want raw features without running inference def get_features(user_ids: list) -> dict: payload = { "dataframe_records": [{"user_id": uid} for uid in user_ids] } resp = requests.post( f"{WORKSPACE_URL}/serving-endpoints/user-features-serving/invocations", headers={ "Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json", }, data=json.dumps(payload), timeout=5, ) return resp.json() # Option 3: Batch scoring (offline) — uses Delta offline store # No online store needed; reads directly from the feature table with PIT lookup batch_labels = spark.table(f"{CATALOG}.gold.users_to_score_today") \ .select("user_id", F.current_timestamp().alias("event_timestamp")) batch_predictions = fe.score_batch( model_uri=f"models:/{CATALOG}.ml.user_churn_model@champion", df=batch_labels, result_type="double", ) batch_predictions.select("user_id", "prediction") \ .write.format("delta").mode("overwrite") \ .saveAsTable(f"{CATALOG}.gold.churn_scores_daily") Feature Table Reference A summary of the feature tables in our pipeline, their update cadence, and their role in the ML lifecycle: Feature TablePrimary KeyTimestamp KeyUpdate MethodLatencyUsed Inuser_activity_featuresuser_idfeature_tsSpark Structured Streaming~5 minReal-time churn, recommendationtransaction_featuresuser_idfeature_tsScheduled batch (hourly)~60 minChurn, LTV predictionuser_profile_featuresuser_idupdated_atCDC from OLTP (near real-time)~2 minAll modelsproduct_featuresproduct_idfeature_tsScheduled batch (daily)~24 hrRecommendation, search rankingsession_featuressession_idsession_end_tsStreaming (micro-batch)~1 minClick-through rate, abandon predictioncohort_featurescohort_idcomputed_atWeekly batch~7 daysSegmentation, A/B analysis Freshness vs cost tradeoff: Streaming features are ~10× more expensive to compute than batch features (continuous cluster vs scheduled job). Only promote a feature to streaming if your model's performance degrades meaningfully with stale data — validate this with an offline ablation study first. Key Takeaways Training-serving skew is the silent killer of production ML — the Feature Store eliminates it by encoding feature computation logic once and using it in both training and serving paths.Point-in-time correct joins via timestamp_lookup_key are non-negotiable for any model trained on time-series data. A missing event_timestamp in your label table is a data leakage bug waiting to happen.fe.log_model() is the right model logging call, not mlflow.sklearn.log_model(). It records feature lineage, enabling reproducible re-training and automatic feature lookup at serving time.Watermarks in Structured Streaming are critical for stateful aggregations — without them, Spark accumulates state indefinitely and the job eventually OOMs. Set them to the maximum tolerable late-data window.Online stores are only worth the operational cost when your SLA is under ~100ms. For batch scoring jobs or APIs with >500ms budgets, read directly from the offline Delta table.fe.score_batch() is the cleanest way to run periodic batch inference — it handles PIT feature lookups automatically, keeps inference logic DRY, and logs results to Delta for downstream consumers. References Databricks — Feature Engineering in Unity Catalog (Overview)Databricks — Create and Manage Online TablesDatabricks — Point-in-Time Feature LookupsApache Spark — Structured Streaming Programming GuideApache Spark — Streaming Watermarks for Late Data HandlingDatabricks — Feature Store Python API ReferenceDatabricks — Score Batch with Feature Store"Feature Stores for ML" — Feast Documentation (open-source reference)"Rethinking Feature Stores" — Chip Huyen (huyenchip.com)Databricks — Model Serving with Automatic Feature Lookup"Building Machine Learning Pipelines" — Hannes Hapke & Catherine Nelson (O'Reilly)

By Jubin Soni, FBCS CORE

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 Soni, FBCS CORE

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.

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.