Big Data

A few days ago, I set out to build a simple image classification model using convolutional neural networks (CNNs). The task itself wasn’t particularly complex, but choosing the right framework proved more challenging than expected. I found myself choosing between TensorFlow and PyTorch, two powerful frameworks for building high-performance CNNs. To explore this, I implemented the same CNN in both frameworks under identical conditions and compared them across key aspects like learning curve, flexibility, debugging, and performance. A Quick Look at the Frameworks Before deep-diving into the comparison, it’s worth briefly understanding the two frameworks used throughout this experiment. 1. TensorFlow TensorFlow is an open-source deep learning framework developed by Google. It is widely known for its strong ecosystem and production-ready capabilities. One of its key strengths is its integration with high-level APIs such as Keras, which simplifies model building and training. TensorFlow is commonly used in large-scale applications, offering tools for deployment across web, mobile, and edge devices. Overall, it is often preferred when moving models from experimentation to production environments. 2. PyTorch PyTorch is an open-source deep learning framework developed by Meta Platforms. It has gained significant popularity, especially in the research community, due to its simplicity and flexibility. PyTorch uses a dynamic computation graph, which makes it feel more like standard Python code. This makes model development more intuitive and debugging significantly easier. It is often the preferred choice for experimentation, rapid prototyping, and research-driven projects. Experiment Setup To ensure a fair and meaningful comparison between TensorFlow and PyTorch, both implementations were designed under identical conditions. 1. Dataset The models were trained and evaluated on the CIFAR-10 dataset, a widely used benchmark for image classification tasks.It consists of 60,000 color images across 10 classes, making it suitable for evaluating CNN performance.CIFAR-10 is publicly available for research purposes and is commonly distributed under a permissive academic license, allowing free use for educational and non-commercial applications. 2. Model Architecture A simple yet effective Convolutional Neural Network (CNN) architecture was used in both frameworks. The structure includes: Convolutional layers for feature extractionReLU activation functionsMax-pooling layers for dimensionality reductionFully connected layers for classification Care was taken to ensure that the architecture remained identical in both implementations. 3. Training Configuration To maintain consistency, the following hyperparameters were used across both frameworks: Optimizer: AdamLearning rate: 0.001Batch size: 64Number of epochs: 10Loss function: Cross-Entropy Loss 4. Environment All experiments were conducted using Google Colab. Both TensorFlow and PyTorch implementations were executed in the same runtime environment. The configuration used includes: Runtime Type: GPU-enabled environmentPython Version: 3.xDeep Learning Libraries: TensorFlow and PyTorch (latest stable versions) The experiments were run on the same Colab runtime session to maintain consistency in resource allocation. Implementation To ensure a fair comparison, the same CNN architecture and training configuration were implemented using both TensorFlow and PyTorch. While the underlying model remains identical, the implementation approach differs significantly across the two frameworks. 1. CNN Implementation in TensorFlow The model was first implemented using TensorFlow with its high-level Keras API, which provides a concise and structured way to define deep learning models. Model Definition Python model = models.Sequential([ layers.Conv2D(32, (3,3), activation='relu', input_shape=(32,32,3)), layers.MaxPooling2D((2,2)), layers.Conv2D(64, (3,3), activation='relu'), layers.MaxPooling2D((2,2)), layers.Flatten(), layers.Dense(64, activation='relu'), layers.Dense(10, activation='softmax') ]) The Sequential API allows layers to be stacked in a linear fashion, making the architecture easy to read and implement. This significantly reduces boilerplate code and is especially helpful for beginners. Model Compilation and Training Python model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy']) history = model.fit(x_train, y_train, epochs=10, batch_size=64, validation_data=(x_test, y_test)) Training in TensorFlow is handled using a single high-level function. It automatically manages the training loop, backpropagation, and metric tracking, making the process highly streamlined. Observation: TensorFlow offers a compact and beginner-friendly implementation. With minimal code, it handles most of the underlying complexity, making it ideal for rapid development and production-oriented workflows. 2. CNN Implementation in PyTorch The same CNN architecture was implemented using PyTorch, which follows a more explicit and flexible approach. Model Definition Python class CNN(nn.Module): def __init__(self): super().__init__() self.conv1 = nn.Conv2d(3, 32, 3) self.pool = nn.MaxPool2d(2,2) self.conv2 = nn.Conv2d(32, 64, 3) self.fc1 = nn.Linear(64*6*6, 64) self.fc2 = nn.Linear(64, 10) In PyTorch, models are defined using Python classes. This provides greater flexibility but requires a more detailed understanding of how each component works. Forward Pass Python def forward(self, x): x = self.pool(torch.relu(self.conv1(x))) x = self.pool(torch.relu(self.conv2(x))) x = x.view(-1, 64*6*6) x = torch.relu(self.fc1(x)) x = self.fc2(x) return x The forward pass must be explicitly defined, giving full control over how data flows through the network. This makes it easier to customize and debug complex models. Training Loop Python for inputs, labels in trainloader: optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() Unlike TensorFlow, PyTorch requires a manual training loop. While this increases the amount of code, it also provides complete transparency and control over the training process. Observation: PyTorch offers a more flexible and transparent approach. Although it requires more code, it allows finer control over model behavior, making it a preferred choice for experimentation and research. With both implementations in place, the next step is to evaluate their performance and analyze how they compare across different metrics. Results and Analysis With both implementations completed under identical conditions, we now compare TensorFlow and PyTorch using empirical results and practical observations. 1. Accuracy The image illustrates the Accuracy and Training Time for TensorFlow and PyTorch. (Image by Author) Both frameworks achieved nearly identical performance on the CIFAR-10 dataset: TensorFlow Accuracy: 68.78% PyTorch Accuracy: 68.95% The difference (0.17%) is extremely small and falls within normal training variation. When architecture, data, and hyperparameters are controlled, the choice of framework has virtually no impact on model accuracy. Additionally, both models show: Consistent improvement across epochsNo signs of severe overfittingStable generalization on test data The image illustrates the Train and Test accuracy for TensorFlow and PyTorch. (Image by Author) 2. Loss Convergence The image illustrates the Loss Convergence for TensorFlow and PyTorch in Logarithmic Scale. (Image by Author) TensorFlow exhibits a smooth and gradually decreasing loss, both for training and validation.PyTorch shows a similar downward trend, but with slightly larger values. The higher loss values in PyTorch are due to loss accumulation across batches, whereas TensorFlow reports average loss per epoch. Despite differences in scale, both frameworks demonstrate stable and consistent convergence behavior, indicating effective training. 3. Model Training Performance Training Speed TensorFlow: 715.23 secondsPyTorch: 723.31 seconds TensorFlow is slightly faster (~1% difference), but the gap is minimal For moderate-sized datasets like CIFAR-10, training speed differences are negligible and unlikely to influence framework selection, but TensorFlow provides strong tooling for large-scale deployment, while PyTorch is equally capable in training large models. 4. Scalability and Flexibility TensorFlow follows a more structured and predefined approach, but provides robust tools such as distributed training and deployment pipelines. It also holds an advantage in large-scale production environments, while PyTorch continues to close the gap. PyTorch uses a dynamic computation graph, allowing runtime modifications, which makes custom modifications easy. It is better suited for research and experimentation, where flexibility is critical. 5. Learning Curve From an implementation standpoint: TensorFlow (via Keras) allows model creation with minimal and structured code; hence, it is easier to start with.PyTorch requires explicit definitions for model architecture, forward passes, and training loops; this results in lengthier code and greater initial effort. Ultimately, the choice between TensorFlow and PyTorch is less about performance and more about how you prefer to design, experiment with, and deploy deep learning models. Choosing Between TensorFlow and PyTorch TensorFlow is better suited when working on production-ready systems, where scalability, deployment tools, and a structured workflow are important. Its high-level APIs make it easy to develop models quickly and integrate them into real-world applications, including mobile and edge environments.PyTorch is more appropriate for research and experimentation, where flexibility and control are critical. Its dynamic nature and seamless debugging experience make it ideal for testing new ideas and building custom architectures. Conclusion: Choosing the Right Framework Through this hands-on comparison of TensorFlow and PyTorch using a CNN on the CIFAR-10 dataset, one key insight becomes clear: both frameworks perform almost identically when it comes to core metrics. The experimental results showed: Nearly identical accuracy (~68–69%)Comparable training timesSimilar loss convergence patterns This highlights an important takeaway: The choice of framework has little to no impact on model performance when architecture and training conditions are kept consistent. However, the real difference lies not in performance, but in how you build, debug, and deploy models. Ultimately, the best framework is not the one that performs slightly better on benchmarks, but the one that aligns with your workflow, problem domain, and development style. Connect with me for more updates: MediumLinkedIN

The Failure You Have Probably Already Seen An enterprise AI agent is deployed against production data. It answers the first ten questions confidently and correctly. Then, on the eleventh question, it produces an answer that looks reasonable but is completely wrong. The team investigates. The model is fine. The prompt is fine. The tool integrations are fine. The problem is buried in the data itself. A field the agent relied on has drifted. A join it assumed existed no longer holds. A quality signal that used to be reliable has silently degraded. This is not a rare edge case. It is becoming one of the most common failure patterns in enterprise AI systems moving from prototype to production. And it points to a simple, uncomfortable truth: most enterprise data infrastructure was built for a consumer we no longer have. I have spent the past couple of years designing agentic AI systems against production data at Fortune 500 scale. What follows is the runtime governance pattern I now design around, and the failure modes it protects against. Who this article is for: This article is for data engineers, platform architects, AI engineers, and governance teams building enterprise agents that depend on production data. It focuses less on prompt design and more on the runtime data controls required to make agent answers reliable. Twenty Years of Data Built for Humans Every large enterprise data platform in production today was designed for human consumption. Analysts, business users, data scientists, and BI teams. Those consumers share a common trait: they exercise judgment. A human analyst looking at a broken dashboard notices it. A data scientist opening a table with unusual distributions asks a colleague. A finance user reviewing a report questions the number when it does not match their gut. Enterprise data governance evolved to support this consumer. Documentation lives in wikis. Quality is enforced by expected-value alerts that a human triages. Lineage is captured at the ETL job level, not the field level. Access is granted through role-based permissions and refined by manual data stewardship. All of this works when a human is at the end of the pipeline. An AI agent is not that consumer. An agent has no judgment. It processes what it is given and returns an answer. If the data is stale, the agent produces a stale answer with high confidence. If the lineage is broken, the agent cannot trace why. If a quality signal exists only as a wiki page, the agent cannot use it. The Four Gaps Most Enterprises Have Across the AI-in-production work I have seen, the same four gaps show up almost every time. Gap 1: Machine-Readable Data Contracts Most contracts exist as documentation, not as programmatic constraints. An agent cannot ask a Confluence page whether it is safe to trust a field. Data contracts need to be enforced at the platform layer, with schema, type, freshness, and quality guarantees expressed as executable rules. Gap 2: Use-Case-Aware Quality Fitness A dataset that is 95 percent complete may be fine for a marketing dashboard and completely wrong for a clinical AI model. Traditional data quality checks are use-case-agnostic. Agentic AI requires quality signals that answer a different question: is this data fit for this specific decision, right now? Gap 3: Field-Level Lineage That Updates in Real Time When a pipeline changes, human consumers get an email. Agents get a wrong answer. Lineage systems need to update as pipelines evolve and expose change signals in a form agents can consume, not just visualize. Gap 4: A Discovery Layer Agents Can Query Most catalog systems are designed for humans to browse. Agents need a machine interface to ask questions like which tables contain the concept I care about, and which of them is authoritative for this domain. Design Principles for Agentic Data Governance Closing these gaps does not require rebuilding the entire data platform. It requires making governance executable in the same path where the agent retrieves data, evaluates context, and produces an answer. Three design principles matter most. Start with the decision, not the data. For each production AI use case, define what a wrong answer looks like and work backward to the data requirements that would prevent it. This surfaces the specific quality signals, lineage nodes, and freshness constraints that matter. Make governance runnable, not readable. Every governance artifact your agents depend on should be programmatically executable at inference time. If a rule cannot be checked in code, an agent cannot use it. Documentation is useful for humans, but for agents it is invisible. Instrument for continuous evaluation. A governance framework that only fires at deployment is not enough. Models drift, data drifts, and use cases evolve. The governance layer needs to continuously evaluate agent outputs against real-world outcomes and flag drift before it becomes damage. Reference Architecture: Runtime Data Governance for AI Agents A practical implementation usually introduces a lightweight runtime governance layer between the agent and the underlying data platform. The goal is not to slow the agent down. The goal is to give the agent a reliable way to ask whether the data behind an answer is safe to use. At a minimum, this pattern includes five components: a data catalog that exposes authoritative sources, a contract registry that stores schema and business rules as executable checks, a lineage service that tracks upstream dependencies at the field and metric level, a quality service that publishes freshness and fitness signals, and an agent guardrail service that evaluates these signals before the agent responds. Runtime flow: User question → Agent → Semantic/data resolver → Governance service → Catalog, contract registry, lineage service, and quality service → Pass/Warn/Block decision → Agent response. Layer Responsibility Example Signal Catalog Identify authoritative datasets and business definitions. Certified source for booked deal value. Contract registry Validate schema, data types, null thresholds, and business rules. Discount variance must use the approved baseline method. Lineage service Track upstream source, transformation, and metric dependencies. Metric changed because a new source was added. Quality service Publish freshness, completeness, anomaly, and fitness scores. Dataset refreshed within SLA and passed threshold checks. Agent guardrail Block, warn, or allow the answer based on governance signals. Answer allowed only if lineage and contract checks pass. The agent should not directly trust a dataset simply because it can access it. Before answering, it should evaluate the data path, the contract status, the freshness window, the lineage change history, and the use-case-specific fitness score. If any critical check fails, the agent should either decline to answer or return the answer with an explicit data reliability warning. How the Runtime Governance Check Works In practice, the check is a short pre-answer step. The agent does not need to understand every governance rule directly. It needs a stable contract with a governance service that can evaluate the data path and return a decision. The user asks a business question.The agent resolves the requested metric, entity, dataset, or semantic concept.The agent calls the governance service with the resolved data assets and intended use case.The governance service checks catalog certification, contract status, lineage changes, freshness, completeness, and use-case fitness.The service returns a pass, warn, or block decision with machine-readable reasons.The agent answers, adds a caveat, escalates, or declines based on that decision. What a Machine-Readable Data Contract Actually Looks Like The abstract idea of a data contract only becomes real when you can point to one that an agent can actually consume. Here is a compact YAML example for a deal variance metric, expressing schema constraints, business rules, freshness expectations, and quality thresholds in a single artifact: YAML contract: dataset: deal.discount_variance schema: - field: discount_variance_pct type: decimal(18,2) required: true calculation: approved_discount_baseline_v2 - field: source_system type: string allowed_values: [crm_v3, revenue_hub] freshness: sla_hours: 24 breach_action: warn quality: completeness_threshold: 0.95 anomaly_score_max: 3.0 lineage: change_window_days: 30 on_upstream_change: require_review With this in place, an agent can call a single governance endpoint before responding, receive a machine-readable pass, warn, or block decision, and either answer confidently, answer with a caveat, or decline. The rule is not buried in a wiki page. It is live at inference time. Example Runtime API Pattern The runtime call does not need to be complicated. A minimal request can identify the metric, dataset, use case, and decision context. The response should be small enough for the agent to use directly in its control flow. JSON POST /governance/evaluate Request: { "metric": "deals.discount_variance_pct", "dataset": "deals.discount_variance", "use_case": "deal_desk_agent_review", "decision_context": "discount_variance_explanation" } Response: { "decision": "warn", "reasons": ["upstream_lineage_changed", "freshness_within_sla"], "agent_action": "answer_with_caveat" } In the agent workflow, this response becomes a control decision. A pass allows the agent to answer normally. A warn allows the answer but requires a reliability caveat. A block prevents the answer and routes the request to review, remediation, or a safer fallback path. Pseudocode: Turning Governance Into Agent Control Flow Python decision = governance.evaluate(metric, dataset, use_case) if decision.status == "block": return decline_with_reason(decision.reasons) if decision.status == "warn": return answer_with_caveat(query, decision.reasons) return answer(query) This is the core shift: governance is no longer a document the team reads during design review. It becomes a runtime dependency that the agent uses to decide whether to answer, qualify the answer, or stop. Runtime Checks an AI Agent Should Perform Before Answering Is this dataset or metric certified for the requested business domain?Has the schema changed since the agent workflow was last validated?Did all required fields meet completeness and validity thresholds?Is the data fresh enough for the decision being requested?Has any upstream lineage changed within a defined risk window?Does the requested answer depend on a metric with multiple calculation methods?Should the agent answer, warn, escalate, or decline based on the governance outcome? This does not require a heavyweight approval workflow for every query. In many cases, the runtime check can be a fast metadata call that returns a simple decision: pass, warn, or block. The important design principle is that governance must be available in the same execution path as the agent response, not in a separate documentation process that only humans can interpret. Failure Modes and Runtime Controls Failure mode What causes it Runtime control Stale answer Dataset missed its refresh SLA. Freshness check with warn or block behavior. Wrong metric Multiple calculation methods exist for the same business concept. Contract and semantic registry validation. Silent lineage change An upstream source or transformation changed after validation. Field-level lineage check within a defined risk window. Misused dataset The dataset is accessible but not certified for the requested domain. Catalog certification and use-case fitness check. Incomplete evidence Required fields fail completeness or validity thresholds. Quality service decision with explicit failure reasons. A Concrete Example From the Field On one enterprise AI project in a regulated environment, we deployed an agentic assistant to help analysts explore a large deal registration and booking dataset. Early testing looked solid. Several weeks into production, the agent began returning confidently wrong answers about a specific discount variance metric. The model had not changed. The prompt had not changed. What changed was an upstream ingestion job that added a new source that computed discount against a different price baseline. A human analyst would likely have questioned the number because it felt off. The agent did not. It saw a valid number in a valid field and reported it as authoritative. The fix was not in the model. We added a machine-readable contract for the approved discount baseline, a lineage signal for recent upstream changes, and a runtime check the agent could call before answering. After that, the same failure could not recur silently. The agent either answered correctly or flagged that the underlying data had changed and required review. The lesson was not that agents are unreliable. It was that agent reliability is a property of the data layer, not the model layer. Once we treated the governance layer as an active runtime dependency instead of static documentation, the entire class of silent-failure risk collapsed. Implementation Considerations Cache low-risk governance decisions to reduce latency, but recheck high-risk metrics at runtime.Separate warn rules from block rules so agents can still answer safely when risk is explainable.Version data contracts alongside pipelines, semantic models, and metric definitions.Log every agent answer with the governance decision, reasons, dataset version, and lineage snapshot used.Start with high-risk metrics and regulated workflows before expanding the pattern across the broader data estate. Why This Belongs in the Architecture, Not the Prompt Prompt engineering can reduce some surface-level errors, but it cannot solve a missing contract, stale dataset, broken lineage path, or ambiguous metric definition. Those failures sit below the model. They need to be handled in the platform architecture, where data access, metadata, quality, lineage, and policy decisions are available at runtime. For teams building enterprise AI agents, the practical takeaway is straightforward: treat runtime governance as part of the agent stack. If an agent can call a retrieval service, vector index, SQL endpoint, or workflow tool, it should also be able to call a governance service before committing to an answer. The next generation of enterprise AI reliability will not come only from better models. It will come from data platforms that can tell agents, in real time, whether an answer is safe to give. About the Author. Avinash Maddineni is a Lead Data Engineer with 15 years of enterprise data infrastructure experience across healthcare, financial services, energy, and travel. He builds agentic AI and data governance systems at Fortune 500 scale and is founder of PureStrokeAI (USPTO provisional patent filed May 2026).

Compliance-reporting teams keep spreadsheets in the loop for a practical reason: a workbook lets domain experts inspect assumptions, formulas, source rows, and intermediate values without reading a line of application code. That transparency is genuinely useful, and it's a big part of why replacing Excel outright so often fails to stick. The trouble starts once that workbook becomes part of a repeatable, audited reporting process — a regulatory filing, an IFRS report, a periodic compliance submission. At that point, a shared Excel file isn't enough on its own. What's actually needed is version control, validation, an audit trail, a review step, and a reliable way to connect the spreadsheet's logic to the systems downstream. The spreadsheet itself isn't the problem. It's a review surface domain experts genuinely need. The problem is treating it as a loose file sitting outside the application. The goal isn't to eliminate spreadsheets, but to preserve the spreadsheet experience while letting the application govern how it's used. This article walks through an architecture that keeps the workbook where domain experts can see it, but moves execution — validation, calculation, output generation, logging — into a Java application. The scenario is inspired by a real-world IFRS reporting project, and the same architecture applies to regulatory reporting, statutory filings, actuarial review, and other spreadsheet-driven compliance workflows. The pattern itself doesn't require a specific product: it works with any spreadsheet engine that can load a workbook and expose read/write access to Java, and parts of it apply even if you only use a file library like Apache POI at the edges. Three Ways Teams Usually Respond Rewrite everything in Java. Engineering gets control, tests, and CI. But the calculation logic moves away from the people who understand it. Every threshold change, every new currency, every adjusted formula now goes through a sprint. Sometimes that's correct — if the rules are stable and nobody inspects formulas, do this. For living, business-owned logic, it breeds shadow spreadsheets. Leave the desktop spreadsheet alone. Finance keeps full flexibility. The organization keeps none of the guarantees: no version control, no audit trail, no way to prove which file produced the submitted numbers. Use a file library only at the edges. Java imports the workbook, exports the results. Better — but the correction loop still happens in desktop Excel: download, fix locally, re-upload, re-validate, repeat. Every round trip is an audit gap. Now there is a fourth option: embed the workbook directly into the web application. Domain experts continue working in a familiar spreadsheet interface, while the application governs when users can edit data, when validation runs, which outputs become visible, and how every operation is logged. The rest of this article is about what that looks like in practice. The Big Idea: One Workbook, Two Roles In this pattern, the workbook plays two roles at the same time: For users, it is the interface. They inspect rows, correct values, maintain rule tables, and review generated outputs in a familiar grid.For the application, it is a runtime artifact. Java loads a known template, reads specific sheets and regions, runs validation, writes outputs, and records every run. The design decision that makes this work: Java never wanders through the workbook looking for data. It reads and writes only through agreed sheets and regions — a contract. Finance owns what's inside the regions: values, formulas, rules. Engineering owns the boundary and everything behind it: execution, permissions, persistence, export. Let's see the two stages of a typical reporting workflow through this lens. Stage 1: Let Users Fix Data Issues Without Leaving the App Reporting source data almost never arrives clean. A currency code says US instead of USD. An FX rate is missing. A service fee breaks a policy limit. The template for this stage has two sheets. Input CSV holds the source rows users can inspect and correct. ETL Rule holds the validation rules — as an ordinary spreadsheet table with columns like Field, Check, and Allowed Values. A rule row might say: currency must be one of USD, EUR. Finance can read and change these rules without asking anyone. When the user clicks Run Validation, the application takes over. To make this concrete: the examples in this article use Keikai Spreadsheet, a Java-based spreadsheet UI component, to embed the workbook in the browser and read and write it from Java — though the same three-step logic applies with any comparable engine. Conceptually, the Java service reads the data rows, reads the rule rows, and checks every row against every rule: Java List<SourceRow> rows = sheetReader.readTable(workbook, "Input CSV"); List<Rule> rules = ruleParser.parse(sheetReader.readTable(workbook, "ETL Rule")); for (SourceRow row : rows) for (Rule rule : rules) rule.check(row).ifPresent(report::add); Notice what this is not: the rules are not hard-coded in Java. Java only knows how to read the rule table and apply generic checks. The actual business knowledge — which currencies are allowed, what a valid fee looks like — stays in the workbook where its owners can see it. One detail carries most of the user experience: every validation error records which cell failed — sheet, row, and column. That lets the UI show a panel saying “policy P-1024, field currency, value US, expected USD or EUR” with a link that jumps the user straight to the offending cell. They fix it in the grid, click run again, and validation passes. Compare that to the traditional loop — download, fix in Excel, upload, pray. Here, nothing leaves the system, and every edit can be logged with user, timestamp, old value, and new value. Stage 2: Generate Outputs Under Application Control Once the data is clean, the second stage produces the actual reporting outputs: journal entries, impact tables, export-ready CSV sheets. The input is a policy sheet with assumptions (premium totals, fees, FX rates) plus a rule table that maps accounting events to journal lines. Before the run, the application shows only the input sheet — output sheets stay hidden, because they don't exist meaningfully yet. When the user triggers generation, the same Keikai-backed workbook is read and written from Java: it reads the inputs, computes the metrics, builds the journal rows, and writes them back into the workbook: Java PolicyInput policy = policyReader.read(workbook, "Policy Input"); Metrics metrics = deriveMetrics(policy); // plain Java arithmetic List<JournalRow> rows = journalBuilder.build(readJournalRules(workbook), metrics); sheetWriter.replaceTable(workbook, "Journal Entries", rows); revealSheets(workbook, "Journal Entries", "Report Impact", "Journal CSV"); The interesting part is the last line. Sheet visibility is an application decision: outputs appear only after a successful run, so a reviewer can never mistake stale output for fresh output. The reviewer then sees everything in one place — assumptions, rules, generated journals, report impact — in the same grid, and the export button produces a file the application has logged and versioned. deriveMetrics itself is deliberately simple — a handful of multiplications and subtractions. In a real system it may be far more complex, or it may even delegate back to formulas in the workbook. The architecture doesn't change: inputs go into agreed regions, outputs come from agreed regions, and Java owns the trigger. The Part Everyone Skips: The Workbook Is Now an API The moment Java code depends on a sheet named ETL Rule with a header called Allowed Values, the workbook has stopped being a document. It has become an interface — and interfaces break when they're changed casually, without review. The fix is to make the contract explicit and test it. Distinguish two kinds of change: Value changes – a new allowed currency, an adjusted threshold, a reviewed formula edit. These live inside the contract. Finance can make them without touching Java.Structural changes – renaming a sheet, deleting a header, moving an output table three columns right. These are API changes and should be reviewed like one. Then write this test: Java @Test void templateSatisfiesReportingContract() { Workbook wb = engine.load("reporting-template.xlsx"); assertSheetExists(wb, "Input CSV", "ETL Rule", "Journal Entries"); assertHeaders(wb, "ETL Rule", "Field", "Check", "Allowed Values"); } It looks almost too simple to matter, but most real-world workbook integration failures are exactly this mundane — a renamed sheet or a deleted header, discovered the night before a regulatory filing is due. Catching it in CI, before any template goes live, is what makes the difference. Finally, log runs, not just files: template version, who ran it, validation status, output row counts, a hash of the inputs. When someone asks “why does this quarter's filing look wrong?”, you answer from the run log instead of from archaeology on a shared drive. For compliance teams, these controls turn the workbook from an informal file into evidence the organization can explain. A reviewer can trace which template version produced a number, which source data was used, who ran the process, whether validation passed, and which output was exported. If a template structure changes, the contract test shows whether the workbook still satisfies the application’s required sheets and headers before it reaches production. In other words, the system does not just calculate results; it records the evidence needed to defend how those results were produced. When to Consider a Simpler Approach This approach pays off when the workbook is a genuine shared language between domain experts and developers — something both sides actually read, edit, and rely on. If the rules rarely or never need to change, and nobody inspects formulas, plain Java is simpler to test and operate. And if the workbook is really just a transfer format between systems, a straightforward import/export covers it. Takeaways The compliance-reporting spreadsheet doesn't have to be rewritten or worked around. Put it inside the application and split ownership along a clear line: The workbook owns what users must see and maintain: source rows, rule tables, assumptions, reviewable outputs.The application owns execution: validation, generation, sheet visibility, permissions, logging, export.The contract between them — named sheets, headers, regions — is documented, tested in CI, and changed only with review. Do that, and the workbook stops being an unversioned file nobody can fully account for. It becomes a governed part of the application — the place where domain experts and the system finally agree on the numbers.

By Hawk Chen 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.

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

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

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

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.