Data

If you have ever tried to build a “mortgage rates today” feature, you have probably landed on the same fork in the road I did. Consumer finance sites publish attractive numbers, but those pages are editorial HTML, not APIs. Layout changes break scrapers, terms of use are unclear, and it is hard to explain what statistic you are actually showing. For a developer project meant to be forked on GitHub, that path is a dead end. I wanted something smaller and more honest: national 30-year and 15-year fixed benchmarks from a source economists already cite, plus a separate surface where first-time buyers could ask basic questions without handing over sensitive data. The result is US Homes Mortgage Agent, an open-source FastAPI application with two clear jobs. One page shows macro rates and a chart. Another page hosts a conversational assistant that is grounded in the same data and in plain Python math. This article explains the problem framing, the architecture, and the implementation choices that kept the project maintainable. The Problem: Rates Pages Are Not Data Products Headline mortgage rates on news sites are useful for humans scanning a story. They are awkward for software. The number in the article may come from a weekly survey, a lender panel, or a marketing table, and the HTML around it changes often. Automating that extraction couples your app to someone else’s front-end design. There is a second trap that is easy to miss. The most common free national series for conventional fixed rates, Freddie Mac’s Primary Mortgage Market Survey, is published weekly through the Federal Reserve Economic Data service (FRED). Series MORTGAGE30US and MORTGAGE15US are the ones you will see referenced in macro commentary. If your UI promises “today’s rate” and refreshes every morning, users will still see the same value for several days. That is correct behavior given the underlying data, but it looks like a bug unless you show the observation date. First-time buyers add a different problem. A chart alone does not answer “what is pre-approval?” or “what is the difference between principal and interest and a full housing payment?” A chat interface helps only if you constrain it. Letting a model invent rates or amortization is worse than no chat at all. What the Application Does The app splits responsibilities instead of blending them into one screen. On the rates and chart view, the server exposes the latest national benchmarks and a history chart with ranges of 7 days, 30 days, 90 days, and one year. Data comes from FRED. A scheduled job runs every day at 9:00 in a configurable timezone (I use America/New_York by default) and stores a snapshot in SQLite. That snapshot powers the headline tiles and feeds one of the assistant’s tools. From the first-time buyer's view, the user gets a simple chat UI. Messages stay in the browser for the session. The server does not write chat history to the database. When the user sends a message, the backend calls an LLM with a system prompt oriented toward education, not legal or tax advice, and with two callable tools: one to read the current benchmark snapshot, and one to compute fixed-rate principal and interest with standard amortization formulas. The product is intentionally not a lender marketplace, not a credit product, and not personalized underwriting advice. It is a reference implementation for macro transparency plus responsible LLM integration. Why FRED, and Why SQLite Only for Benchmarks FRED gives documented time series with a free API key. For this project, that was the right trade: legal programmatic access, stable identifiers, and enough history for charts without building an ETL pipeline on day one. The SQLite database holds one concern: daily benchmark snapshots. Each successful refresh upserts a row keyed by calendar date with 30-year and 15-year values, the FRED observation dates for each series, and a fetched_at timestamp. Chart data for longer windows is fetched from FRED on demand when the user changes the range. I accepted extra API calls on chart loads in exchange for not maintaining a large local history store. Chat content never lands in that schema. If you are cloning the repo, your local data/ directory is git ignored along with .env, so you start fresh without accidentally publishing keys or a test database. Architecture at a Glance The deployment model is deliberately boring: one FastAPI process, static assets, Jinja templates, and an in-process APScheduler cron job. That keeps the project easy to run locally with uvicorn and easy to reason about for readers who want to fork it. Logical component diagram: browser, FastAPI, SQLite, FRED API, and LLM HTTP API At runtime, three paths matter. Page load for rates. The browser requests HTML, then calls /api/rates/today for the cached snapshot and /api/rates/chart for series data. If FRED is slow on a 90-day or one-year range, the UI shows a loading state and disables the range control so users know the chart is working. Scheduled refresh. At 9:00 local time, the job pulls the latest observations for both series and upserts the snapshot. Between weekly FRED releases, the values may not change. The UI still benefits from a daily pull because you pick up new observations as soon as they exist. Assistant chat. The client posts the in-memory conversation to /api/chat. The server prepends a system prompt that includes the current benchmark context, calls the LLM with tool definitions, and loops if the model requests tool execution. When tools return JSON, the model produces the final natural-language answer. Grounding the LLM With Tools The pattern I cared about most was keeping numbers out of the model’s imagination. The assistant exposes two tools. get_benchmark_rates reads the same SQLite snapshot as the dashboard tiles and returns structured JSON with rates and observation dates. estimate_monthly_pi takes the loan amount, annual rate, and term, then runs deterministic amortization in Python. Taxes, insurance, and PMI are out of scope for that calculation, and the prompt tells the model to say so. Here is the core amortization helper. It is small on purpose: easy to test, easy to audit. def monthly_principal_interest(loan_amount, annual_rate_percent, years): n = years * 12 r = (annual_rate_percent / 100.0) / 12.0 if r == 0: return round(loan_amount / n, 2) payment = loan_amount * (r * (1 + r) ** n) / ((1 + r) ** n - 1) return round(payment, 2) The system prompt reinforces guardrails: stay educational, refuse to handle sensitive identifiers like SSNs, and point users toward HUD counselors or licensed lenders for personal decisions. That does not replace compliance review if you productize this, but it sets a baseline for an open demo. LLM Provider Choice The reference code uses OpenAI’s Chat Completions API with OPENAI_API_KEY and defaults the model to gpt-4o-mini. The HTTP client posts to OPENAI_BASE_URL, which defaults to https://api.openai.com/v1. In practice, any host that implements a compatible chat completions endpoint and supports tool calling can be configured the same way, including several OpenAI-compatible gateways. You should verify tool-call behavior and error formats on your chosen provider before relying on it in production. The rates dashboard works with only FRED_API_KEY. The assistant returns a clear configuration error until an LLM key is present, which makes local development predictable. Operations and Open Source Hygiene For a public GitHub repository, secrets stay in .env. FRED and LLM keys are required at runtime but must not be committed. SQLite files live under data/ and are ignored by git. If you deploy behind a reverse proxy, terminate TLS there and probe /api/health. One caveat for horizontal scaling: APScheduler runs inside each process. Multiple replicas without coordination will each run the 9:00 job. For a demo, that is harmless. For production, you would externalize scheduling or elect a single worker for the cron task. LLM calls have cost and retention implications on the provider side. Even though this app does not store chats server-side, the provider still processes prompt content under its own policies. Plan for that if you expose the assistant publicly. What I Would Extend Next National weekly averages are the right free starting point, not the final word. State-level pricing, daily lender indices, and streaming chat responses are all reasonable extensions, each with its own licensing or infrastructure cost. Automated tests against mocked FRED responses and golden tests for the amortization helper would be the first engineering upgrades I would prioritize. Closing Thoughts The useful analysis from this build is the separation of concerns. Macro data comes from FRED with explicit observation dates. Education flows through an LLM that is allowed to explain and summarize, not to invent rates. Math runs in code the way it always should have. If you want to explore the implementation, the project is open source. Clone it, add your own API keys, and treat it as a starting point rather than a finished financial product.

Apache Parquet became the default format for analytical data because it matched the read path of analytical engines. Queries scanned large parts of a dataset, often across a small set of columns, and Parquet was built to support that efficiently. Row groups, column pages, and compression all work well when the goal is to maximize scan throughput. That model still fits a large part of analytics. But it starts to break down when queries read small subsets of data, especially when those reads are repeated. At that point, the cost is no longer dominated by scanning. It depends on how much data the reader must process before it can return the result. That is where comparing Parquet with Lance becomes useful; the difference is not just in file format, but in the read path itself. The Lance paper frames this problem well by focusing on how structural encoding affects random access and scan performance. Running the Examples Locally All of the examples below can run on a laptop. Install the dependencies with: pip install pandas pyarrow pylance numpy The Python package is pylance, but it is imported as lance. The official Lance docs and Python SDK docs are useful if you want to explore the API surface further. If you are using Homebrew Python on macOS and see an externally-managed-environment error, use a virtual environment instead: Shell python3 -m venv parquet-lance-demo source parquet-lance-demo/bin/activate pip install pandas pyarrow pylance numpy Where the Difference Starts Parquet and Lance are both columnar formats, but they are optimized around different kinds of access. Parquet is built around scan-heavy workloads. Data is typically written once, stored in larger chunks, and read sequentially. That design improves compression and makes it easier for analytical engines to process large volumes of data efficiently. Lance takes a different path. It is designed for workloads where queries may repeatedly touch small parts of a dataset, where latency matters more, and where similarity search is part of the data access path rather than an external system layered on top. This difference is easiest to understand with two concrete examples. The first is a selective filter. The second is vector similarity search. Three Read Paths to Keep in Mind The easiest way to compare Parquet and Lance is to start with the read path. A scan-oriented read path is the classic analytical case. The query reads a meaningful portion of a dataset, usually across a subset of columns. Parquet performs well here because the reader can process row groups and column pages efficiently. A selective read path behaves differently. The query may return only a few rows, but the reader still needs to identify where those rows live. If the format works mainly at chunk granularity, the system may read and decode more data than it returns. A vector-native read path is different again. The query is not asking whether a row satisfies a predicate like id < 100. It is asking which rows are closest to a query vector. That requires an index-aware retrieval path, not only a scan path. Use Case 1: Selective Reads Start with something familiar: a basic filter. WHERE id < 100 On a dataset with millions of rows, this returns almost nothing. The interesting part is not the result. The interesting part is how much work the system performs before it gets there. To make that visible, I used the same benchmark on both formats. The script below generates a dataset, writes it to Parquet and Lance, and then runs the same filter several times. Python import os import shutil import time import numpy as np import pandas as pd import pyarrow as pa import lance # Try 1M, 5M, 10M to observe scaling behavior N = 5_000_000 # Small result set -> stresses selective access vs scan FILTER_EXPR = "id < 100" # Run multiple times; use best to reduce noise RUNS = 5 def clean_outputs(): if os.path.exists("data.parquet"): os.remove("data.parquet") if os.path.exists("data.lance"): shutil.rmtree("data.lance") def time_it(name, fn, runs=RUNS): times = [] result = None for _ in range(runs): start = time.time() result = fn() times.append(time.time() - start) print(f"{name} times:", times) print(f"{name} best:", min(times)) return result def main(): clean_outputs() print(f"Generating dataset with {N} rows...") df = pd.DataFrame({ "id": np.arange(N), "value": np.random.rand(N) }) # Parquet write (scan-optimized) start = time.time() df.to_parquet("data.parquet") print("Parquet write time:", time.time() - start) # Lance write (Arrow-based) start = time.time() table = pa.Table.from_pandas(df) lance.write_dataset(table, "data.lance", mode="overwrite") print("Lance write time:", time.time() - start) dataset = lance.dataset("data.lance") # Selective filter: returns ~100 rows result_parquet = time_it( "Parquet filter", lambda: pd.read_parquet("data.parquet").query(FILTER_EXPR) ) result_lance = time_it( "Lance filter", lambda: dataset.to_table(filter=FILTER_EXPR).to_pandas() ) print("Parquet rows:", len(result_parquet)) print("Lance rows:", len(result_lance)) if __name__ == "__main__": main() Here is a sample run from my laptop using 5 million rows: Shell (parquet-lance-demo) hitarth@hitarth % python parquet_vs_lance.py Generating dataset with 5000000 rows... Parquet write time: 0.14815711975097656 Lance write time: 0.09784913063049316 Parquet filter times: [0.09580063819885254, 0.04515504837036133, 0.03702282905578613, 0.04055309295654297, 0.03741908073425293] Parquet filter best: 0.03702282905578613 Lance filter times: [0.0851907730102539, 0.012360095977783203, 0.009278059005737305, 0.008661031723022461, 0.007877826690673828] Lance filter best: 0.007877826690673828 Parquet rows: 100 Lance rows: 100 The first Lance run was slower than the rest, but repeated runs stabilized quickly. The same pattern showed up across larger dataset sizes as well. I also ran the benchmark at 1 million, 5 million, and 10 million rows. In all cases, the query returned 100 rows. These were the best-of-five timings: dataset sizeparquet filterlance filter 1,000,000 rows 0.033s 0.010s 5,000,000 rows 0.037s 0.008s 10,000,000 rows 0.073s 0.016s The numbers matter less than the pattern. The result size stays constant, but Parquet’s time increases with dataset size while Lance remains relatively stable after the initial read. That points directly to a difference in the read path. On a laptop, this difference shows up as milliseconds. In a production data lake, the same pattern can become more expensive. Extra chunks do not only mean extra CPU. They can also mean additional object-store reads, decompression work, memory materialization, and network latency. A selective query over Parquet may return a tiny result set, but still pay part of the cost of scanning and decoding larger units of data. That is the practical form of read amplification. A compact way to visualize it is this: Why Parquet Takes This Path Parquet is not just a file of columns. Internally, a file is organized into row groups; each row group contains one column chunk per column, and column chunks are divided into pages, as described in the Parquet concepts and file format documentation. Parquet metadata also helps the reader skip some work before decoding begins, which is one of the format’s core strengths. See the metadata documentation. I covered the Parquet scan path in more detail in my earlier DZone article, Understanding Parquet Scans, so I’ll keep this recap focused on what changes in the read path as the workload becomes more selective. Plain Text [ repetition levels ] [ definition levels ] [ values ] Those pages do not hold only values. They also hold structural information needed to reconstruct rows, especially when nested data is involved. This is also the lens used in the Lance paper. It argues that structural encoding, especially repetition, validity, and page layout, has a direct impact on random-access cost, read amplification, and decode overhead. When a reader evaluates a predicate over Parquet data, it first uses metadata to decide which row groups may be relevant. It can skip some work at that level, which is one of Parquet’s strengths. But once a row group has been selected, the reader still needs to read column pages, decode them, reconstruct the row structure, and only then evaluate the filter. That path is efficient when a query is scanning a large fraction of the dataset. It is less efficient when the result is tiny. The smallest useful unit of work is still a chunk. This is why selective queries can feel disproportionately expensive in Parquet. The format is doing exactly what it was designed to do. It is just optimized around chunk-level processing rather than lookup-oriented access. What Changes in Lance Lance changes that path earlier. Instead of treating most queries as scan-first, Lance uses dataset metadata and access structures to narrow the read before reconstruction begins. The official read and write guide is a good starting point for the dataset API used in the examples below. The reader can identify relevant fragments, read only the necessary data, and return results without paying the same chunk-level decode cost across the rest of the dataset. For selective reads, the practical effect is that the reader can identify relevant fragments and avoid decoding larger portions of the dataset. For vector workloads, the index becomes even more important because the query is not looking for an exact predicate match. It is looking for nearby vectors. That is the practical meaning behind the benchmark. The query returns 100 rows in both cases, but the amount of data processed before those rows are produced is different. This distinction becomes clearer as the dataset grows because Parquet’s work still tracks chunk boundaries while Lance’s work is tied more closely to the size of the result. Use Case 2: Vector Similarity Filtered reads are still part of traditional analytics. Vector search is a different kind of workload. A vector is just a list of numbers that represents something like text, an image, or a user. Instead of filtering by exact values, a system compares vectors and returns the nearest matches. A traditional query looks for rows that satisfy a predicate. A vector query looks for rows that are similar. In larger systems, vector search is usually implemented with an approximate nearest neighbor index, often called an ANN index. The index avoids comparing the query vector against every stored vector. Instead, it narrows the search to candidates that are likely to be close. This trades a small amount of exactness for much faster retrieval. In many data lake architectures, this index lives outside the analytical dataset. Vectors may be stored in Parquet, then copied into a separate vector database or indexing service. That creates another pipeline to maintain and another consistency problem to manage. This is why vector-native storage matters. The important change is not only that vectors can be stored. It is that retrieval becomes part of the dataset access path. That difference sounds abstract until you tie it to something familiar. This shows up in semantic search, recommendations, LLM retrieval, and image similarity. In each case, the system is not looking for an exact value. It is looking for nearby representations. The shape of the query changes from this: WHERE id = 42 → exact match to this: query → vector → nearest neighbors That changes what the storage layer needs to support. Here is the benchmark I used for the vector case: Python import os import shutil import time import numpy as np import pandas as pd import pyarrow as pa import lance # Try 1M, 5M, 10M to observe scaling behavior N = 5_000_000 # Small result set -> stresses selective access vs scan FILTER_EXPR = "id < 100" # Run multiple times; use best to reduce noise RUNS = 5 def clean_outputs(): if os.path.exists("data.parquet"): os.remove("data.parquet") if os.path.exists("data.lance"): shutil.rmtree("data.lance") def time_it(name, fn, runs=RUNS): times = [] result = None for _ in range(runs): start = time.time() result = fn() times.append(time.time() - start) print(f"{name} times:", times) print(f"{name} best:", min(times)) return result def main(): clean_outputs() print(f"Generating dataset with {N} rows...") df = pd.DataFrame({ "id": np.arange(N), "value": np.random.rand(N) }) # Parquet write (scan-optimized) start = time.time() df.to_parquet("data.parquet") print("Parquet write time:", time.time() - start) # Lance write (Arrow-based) start = time.time() table = pa.Table.from_pandas(df) lance.write_dataset(table, "data.lance", mode="overwrite") print("Lance write time:", time.time() - start) dataset = lance.dataset("data.lance") # Selective filter: returns ~100 rows result_parquet = time_it( "Parquet filter", lambda: pd.read_parquet("data.parquet").query(FILTER_EXPR) ) result_lance = time_it( "Lance filter", lambda: dataset.to_table(filter=FILTER_EXPR).to_pandas() ) print("Parquet rows:", len(result_parquet)) print("Lance rows:", len(result_lance)) if __name__ == "__main__": main() Here is a sample run from my laptop: Shell (parquet-lance-demo) hitarth@hitarth parquet_lance % python vector.py Generating 100000 vectors of dimension 128... Lance write time: 0.07955217361450195 Vector search time: 0.13872408866882324 id vector _distance 0 50506 [0.053919002, 0.36111426, 0.2145877, 0.9197419... 12.631166 1 41428 [0.17633885, 0.71251565, 0.072742924, 0.759959... 12.962575 2 3216 [0.06450701, 0.24716537, 0.41617322, 0.624773,... 13.008584 3 50216 [0.13460344, 0.9618073, 0.8334099, 0.56230646,... 13.097234 4 75019 [0.35094073, 0.11819457, 0.44928855, 0.0426102... 13.124901 This wrote 100,000 vectors of dimension 128 and returned the top 5 nearest vectors in about 139 ms. The exact number is less important than the query path itself: the search runs directly against the dataset and returns nearest neighbors with distances. Parquet can store the same vector column, but it does not provide a native nearest-neighbor query path. In practice, the workflow usually looks like this: Parquet → extract vectors → build index → query With Lance, the storage layer participates directly in the query: dataset → query directly That is not just a performance difference. It is a capability difference. Lance’s official documentation also exposes SDK-level APIs for working with datasets and vector-oriented workflows through the SDK docs. Lance expects vector columns as fixed-size arrays rather than generic variable-length Python lists. That lets the system reason about dimensionality during query execution. In other words, the structure of the stored data is part of making the query possible. Tradeoffs and Operational Considerations This comparison should not be read as "Lance replaces Parquet." The two formats are useful in different parts of a data platform. Parquet remains the safer default for broad analytical workloads. It has mature support across query engines, data lakes, catalogs, ingestion systems, and governance tooling. If the workload is mostly batch analytics, reporting, or large aggregations, Parquet’s scan-oriented design is still a very good fit. Lance becomes interesting when the workload starts to depend on repeated selective access, lower-latency retrieval, or vector-native queries. In those cases, avoiding unnecessary decoding or avoiding a separate vector indexing pipeline can matter more than raw scan throughput. A simplified comparison looks like this: areaparquetlance Best fit Large analytical scans Selective reads and vector retrieval Read path Row groups and pages Fragment and index-aware access Predicate filtering Metadata and page-level pruning Metadata/index-assisted narrowing Vector search Usually external system Native query path Ecosystem maturity Very mature Emerging Engine compatibility Broad support across engines Narrower ecosystem Updates Usually rewrite and compact Dataset-level mutation support Operational default Strong default for data lakes Better fit for specialized access patterns The operational question is not which format is generally better. The better question is where the workload spends its time. If most queries scan large portions of data, Parquet is still the right default. If the workload repeatedly asks for a small number of rows or nearest neighbors, a lookup-oriented or vector-native format becomes more attractive. What the Two Examples Show Together The filtered read example and the vector example highlight two different consequences of the same design choice. In the filtered read case, the difference appears as execution cost. Both systems can answer the query, but the amount of data processed before returning the result is different. In the vector case, the difference appears as capability. One format stores the data, while the other format also provides a native query path over it. Both cases come back to the same question: how much data must the system process before it can produce the answer? For scan-heavy analytics, Parquet remains a strong fit. That is the read path it was built for. But when workloads shift toward selective access, repeated reads, or similarity search, the main question changes. The difference is no longer just how fast data can be scanned. It is how much data the reader must process before returning the result. That is the broader storage trend. File formats are becoming more involved in the read path itself. They increasingly encode assumptions about access patterns, indexing, and retrieval. As analytical, ML, and search workloads move closer together, storage layout becomes part of query design.

The adoption of retrieval-augmented generation (RAG) from research papers to production systems has been rapid. Those who tried it in 2023 are now deploying it at scale for enterprise search, internal knowledge bases, and customer-facing assistants. However, a lot is still between a working prototype RAG and one that can withstand traffic on the road, using real data, and real modes of failure. This article explains what this gap is, how to plug it, and where most production pipelines fail. What a Production RAG Pipeline Involves A basic RAG pipeline consists of three components: a document store, a vector database to store embeddings, and a language model to produce answers based on the retrieved context. There, most of the tutorials end. Production systems have no such convenience. A production pipeline also needs: Ingestion pipelines that process document updates, deletions, and format variationsSemantic preserving chunking strategies for different document types.Incorporating models that align with the retrieval use case (asymmetric vs. symmetric search).A retrieval layer to support hybrid search, metadata filtering, and re-rankingA generation layer that has prompt management, output validation, and fallback behaviorEvaluation is hooked at each stage so that one can catch the degradation in time for the users It is easy to make each piece individually. The engineering challenge is to get them to cooperate under production conditions. Choosing the Right Vector Database Everything downstream is influenced by the choice of the vector DB. The top choices for 2026 are Pinecone, Weaviate, Qdrant, Milvus, and (for teams already using Postgres) pgvector. A few things that matter more than raw benchmark performance: Filtering support. If your documents have metadata (date, category, author, department), then you require a vector DB that filters pre-retrieval and not post-retrieval. Recall precision is destroyed by post-retrieval filtering. Both Qdrant and Weaviate do a good job of this. In recent versions, Pinecone introduced support for metadata filters, though there are edge cases for large sets of filters. Update semantics. Some vector DBs are delete-and-insert, which means there may be retrieval gaps while they are reindexing. In some cases, such as summarizing news or for internal documentation, this behavior can result in stale or missing results. Compared to most, Milvus performs rolling updates more gracefully. Hybrid search. Dense retrieval would not find exact matches for keywords. Sparse retrieval (BM25-style) ignores semantic similarity. Most production teams ultimately choose to use both, employing a fusion layer (typically the default Reciprocal Rank Fusion). Weaviate has this integrated. For others, it is wired up by yourself. Scale and cost. pgvector performs well up to ~1-2 million vectors on reasonable hardware. In addition, a specialized vector database pays for itself. Chunking Strategy: The Part Most Teams Get Wrong The way you partition documents to be embedded affects the quality of retrieval more than any other choice. The simplest method is fixed-size chunking with a token limit and yields fairly poor results for a structured document such as a PDF, a contract, or technical specifications. Better approaches: Recursive character splitting with overlap preserves context across chunk boundaries. LangChain's RecursiveCharacterTextSplitter is a typical first step, with chunk sizes of 512-1024 tokens and an overlap of 10-15%. Semantic chunking is similar to grouping sentences by embedding similarity (not token count). This is more suitable for conversational transcripts and prose-heavy documents. The downside is that it takes longer to calculate when ingesting. Document-aware chunking parses the structure before splitting. If HTML or Markdown, you split on headers. In the case of PDFs, you use layout analysis to separate tables, figures, and body text. Libraries like unstructured.io handle much of this automatically. A pattern that should be adopted early: Keep the entire document with the chunk. At the time of retrieval, you retrieve the chunk to score for relevance, and if needed, you pull some of the context from the parent document to generate. This is known as a parent-child retrieval pattern, and is a great way to get higher quality answers to longer context queries without increasing the size of your index. Re-Ranking: The Layer Most Prototypes Skip Vector similarity search returns the top-k most similar chunks. That's not necessarily the top-k most useful chunks that answer a specific query. Instead of scoring each individual piece of information, cross-encoder re-rankers process the query and the information to be retrieved as pairs and return a relevance score that considers the relationship. Models like cross-encoder/ms-marco-MiniLM-L-6-v2 are small enough to run in real-time and measurably improve precision on mixed-topic corpora. Agentic RAG setups take this one step further in 2026. The agent analyzes the results retrieved, and if they are not enough, the agent will perform the retrieval again. This iterative pattern is for multi-hop questions that can't be answered by a single retrieval round. For enterprise knowledge applications with a focus on the quality of the answer, rather than latency, teams developing with LangGraph or CrewAI are turning to this pattern. Prompt Management and Output Validation The generation layer fails in expected ways: it hallucinates when the retrieved context is irrelevant, it hedges when there is conflicting retrieved context, or it simply ignores the context and generates from training data. There is no universal solution to each failure mode. Add a faithfulness check to the hallucination. Once the generation is complete, have the LLM perform a separate call to check if the answer is based on the retrieved passages or not. This is a metric that is provided by tools such as RAGAS and can be tracked over time. Don't hardcode prompts in application code for prompt management. Having the ability to perform rapid iterations of a prompt without redeployment, and identify which iteration was active at the time of an incident (a prompt registry, even a basic one in a config file or database) is a huge plus. Evaluation: The Missing Foundation The majority of RAG pipelines are delivered without any systematic evaluation. The teams that end up in trouble 6 months later are the ones that have neglected to do this. The minimum viable evaluation setup includes: Golden Data Set: 100-200 representative question-answer pairs from real use.Metrics for retrieval (recall@k, MRR) and generation (faithfulness, answer relevance, context precision)A nightly or weekly eval run that compares current performance against a baseline Both RAGAS and TruLens connect to popular RAG frameworks and provide you with the above metrics without having to build them yourself. Where Teams Often Need Outside Help When creating RAG pipelines for the first time, teams often overlook the operational aspects, such as implementing model versioning, handling migrations of indexes when switching models, and tracking retrieval drift as document corpora expand. As companies transition to production-level generative AI development, the combination of in-house engineering and generative AI consulting can rapidly address these gaps before hiring and training all the necessary personnel. It is where generative AI integration services can contribute the most value: from architecture to production, running and monitored systems. When getting started on the selection process for a vector DB, the chunking strategy, and evaluation infrastructure, ensure that you have the correct setup for a production RAG system. Those decisions will be much easier to get right from the start rather than after your index is up and running.

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 Abhishek Soni 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.

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 Abhishek Soni CORE

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.

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

By Jubin Abhishek Soni CORE

Why Fine-Tune on Databricks? General-purpose LLMs like Llama 3, Mistral, or Falcon are impressive out of the box — but they underperform on domain-specific tasks: medical coding, legal clause extraction, internal support ticket classification, and financial report summarization. Fine-tuning adapts a pre-trained model's weights to your domain using your proprietary labeled data. Doing this at scale introduces real engineering challenges: Training data lives in Delta Lake across dozens of tablesGPU clusters need to be orchestrated, not hand-managedExperiment tracking must be reproducible and auditableModels need a promotion workflow before they touch production traffic Databricks solves all of this in one platform: Apache Spark for large-scale data preparationMLflow (built-in) for experiment tracking, model registry, and lineageDatabricks Model Serving for one-click deployment with auto-scalingUnity Catalog for governed model and data access The ML Lifecycle Architecture Training Pipeline: End-to-End Flow The flow below shows how a single training run moves through the system — from a triggered job to a promoted model alias. Environment Setup Python # Databricks Runtime ML 14.x+ recommended (ships CUDA, PyTorch, Transformers) # Install additional packages in your cluster init script or notebook %pip install \ transformers==4.40.0 \ peft==0.10.0 \ trl==0.8.6 \ accelerate==0.29.3 \ horovod[spark]==0.28.1 \ datasets==2.19.0 \ evaluate==0.4.1 \ --quiet dbutils.library.restartPython() import os import mlflow import mlflow.transformers import torch from transformers import ( AutoTokenizer, AutoModelForCausalLM, TrainingArguments, Trainer, DataCollatorForLanguageModeling, ) from peft import LoraConfig, get_peft_model, TaskType from pyspark.sql import functions as F from datasets import Dataset # ── MLflow setup ────────────────────────────────────────────────────────────── # On Databricks, MLflow tracking URI is pre-configured to the workspace # mlflow.set_tracking_uri("databricks") # uncomment for external clusters EXPERIMENT_NAME = "/Users/[email protected]/llm-finetuning/support-classifier" mlflow.set_experiment(EXPERIMENT_NAME) BASE_MODEL = "mistralai/Mistral-7B-Instruct-v0.2" CATALOG = "prod" GOLD_DB = f"{CATALOG}.gold" MODEL_NAME = f"{CATALOG}.ml.support_intent_classifier" # Unity Catalog model path print(f"GPU available: {torch.cuda.is_available()}") print(f"Device count: {torch.cuda.device_count()}") Preparing Training Data With Spark Spark handles the heavy lifting before training: filtering noisy records, formatting prompt-response pairs, and splitting the dataset. This stage runs on the CPU cluster — GPU nodes only spin up for the actual training job. Plain Text # ── Spark Data Preparation ──────────────────────────────────────────────────── def build_prompt(row): """ Format a support conversation into an instruction-following prompt. Uses the Mistral instruct template: [INST] ... [/INST] """ return f"[INST] Classify the intent of this support message:\n\n{row['message']} [/INST] {row['intent_label']}" # Load from Delta Gold table raw_df = ( spark.table(f"{GOLD_DB}.support_conversations") .filter(F.col("quality_score") >= 0.85) # keep high-quality labels only .filter(F.col("intent_label").isNotNull()) .filter(F.length("message") > 20) # filter empty/stub messages .filter(F.length("message") < 2048) # filter messages too long to tokenize .dropDuplicates(["message_hash"]) # remove exact duplicates .select("message", "intent_label", "created_date") .limit(500_000) # cap for this training run ) print(f"Training candidates: {raw_df.count():,}") # Build prompt strings using Spark — parallelized across all workers prompt_udf = F.udf( lambda msg, label: f"[INST] Classify the intent of this support message:\n\n{msg} [/INST] {label}", returnType="string" ) prepared_df = ( raw_df .withColumn("prompt", prompt_udf(F.col("message"), F.col("intent_label"))) .withColumn("token_count", F.size(F.split(F.col("prompt"), r"\s+"))) # rough word count proxy .filter(F.col("token_count") < 512) # stay within model context .select("prompt", "token_count", "created_date") ) # Stratified split using Spark (reproducible with seed) train_df, val_df, test_df = prepared_df.randomSplit([0.80, 0.10, 0.10], seed=42) # Persist splits to Delta for lineage + reproducibility train_df.write.format("delta").mode("overwrite").saveAsTable(f"{GOLD_DB}.llm_train_split") val_df.write.format("delta").mode("overwrite").saveAsTable(f"{GOLD_DB}.llm_val_split") test_df.write.format("delta").mode("overwrite").saveAsTable(f"{GOLD_DB}.llm_test_split") print(f"Train: {train_df.count():,} | Val: {val_df.count():,} | Test: {test_df.count():,}") Fine-Tuning With Hugging Face + MLflow Tracking We use LoRA (Low-Rank Adaptation) — a parameter-efficient fine-tuning technique that freezes the base model and only trains a small set of adapter matrices. This cuts GPU memory requirements by ~70% compared to full fine-tuning, making 7B parameter models trainable on a single A100. Python # ── LoRA Fine-Tuning with MLflow Autolog ───────────────────────────────────── # Convert Spark DataFrame to Hugging Face Dataset train_pd = spark.table(f"{GOLD_DB}.llm_train_split").select("prompt").toPandas() val_pd = spark.table(f"{GOLD_DB}.llm_val_split").select("prompt").toPandas() hf_train = Dataset.from_pandas(train_pd) hf_val = Dataset.from_pandas(val_pd) # Load tokenizer and base model tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL, padding_side="right") tokenizer.pad_token = tokenizer.eos_token def tokenize(batch): return tokenizer( batch["prompt"], truncation=True, max_length=512, padding="max_length", ) hf_train_tok = hf_train.map(tokenize, batched=True, remove_columns=["prompt"]) hf_val_tok = hf_val.map(tokenize, batched=True, remove_columns=["prompt"]) # Load base model in 4-bit quantization (QLoRA) from transformers import BitsAndBytesConfig bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_use_double_quant=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.bfloat16, ) base_model = AutoModelForCausalLM.from_pretrained( BASE_MODEL, quantization_config=bnb_config, device_map="auto", trust_remote_code=True, ) # Apply LoRA adapter config lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=16, # rank — higher = more capacity, more memory lora_alpha=32, # scaling factor lora_dropout=0.05, target_modules=["q_proj", "v_proj"], # attention layers to adapt bias="none", ) model = get_peft_model(base_model, lora_config) model.print_trainable_parameters() # Typical output: trainable params: 13,631,488 || all params: 3,765,522,432 || trainable: 0.36% # Training arguments training_args = TrainingArguments( output_dir="/dbfs/tmp/llm-finetune/checkpoints", num_train_epochs=3, per_device_train_batch_size=4, per_device_eval_batch_size=4, gradient_accumulation_steps=8, # effective batch size = 32 warmup_ratio=0.03, learning_rate=2e-4, fp16=False, bf16=True, # use bfloat16 on A100/H100 logging_steps=50, eval_strategy="steps", eval_steps=200, save_strategy="steps", save_steps=200, load_best_model_at_end=True, metric_for_best_model="eval_loss", report_to="mlflow", # pipe all metrics to MLflow automatically ) data_collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False) trainer = Trainer( model=model, args=training_args, train_dataset=hf_train_tok, eval_dataset=hf_val_tok, tokenizer=tokenizer, data_collator=data_collator, ) # ── MLflow Run ──────────────────────────────────────────────────────────────── with mlflow.start_run(run_name="mistral-7b-lora-v1") as run: # Log hyperparameters manually for full auditability mlflow.log_params({ "base_model": BASE_MODEL, "lora_rank": lora_config.r, "lora_alpha": lora_config.lora_alpha, "lora_dropout": lora_config.lora_dropout, "target_modules": str(lora_config.target_modules), "quantization": "4-bit QLoRA (nf4)", "train_samples": len(hf_train_tok), "val_samples": len(hf_val_tok), "epochs": training_args.num_train_epochs, "effective_batch": training_args.per_device_train_batch_size * training_args.gradient_accumulation_steps, "learning_rate": training_args.learning_rate, }) # Train — metrics auto-logged to MLflow via report_to="mlflow" trainer.train() # Log final eval metrics explicitly eval_results = trainer.evaluate() mlflow.log_metrics({ "final_eval_loss": eval_results["eval_loss"], "final_eval_perplexity": torch.exp(torch.tensor(eval_results["eval_loss"])).item(), }) # Log the model + tokenizer as a single MLflow artifact mlflow.transformers.log_model( transformers_model={"model": trainer.model, "tokenizer": tokenizer}, artifact_path="model", task="text-generation", registered_model_name=MODEL_NAME, # auto-registers to Unity Catalog metadata={"base_model": BASE_MODEL, "finetuning": "QLoRA"}, ) run_id = run.info.run_id print(f"Run ID: {run_id}") print(f"Eval Loss: {eval_results['eval_loss']:.4f}") Distributed Training With Horovod on Spark For datasets beyond a few million tokens, or when you need to fine-tune models larger than 13B parameters, single-node training hits GPU memory walls. Horovod distributes training across multiple GPU workers using ring-allreduce — each worker holds a full model replica, and gradients are averaged across workers after every backward pass. Python # ── Distributed Fine-Tuning with Horovod on Spark ──────────────────────────── # Best for: datasets > 5M tokens, models > 13B params, or when you need # to reduce wall-clock training time below a business SLA. import horovod.torch as hvd from sparkdl import HorovodRunner def train_fn(hparams): """ Training function executed on each Horovod worker. Each worker trains on a data shard; gradients are averaged across workers. """ import horovod.torch as hvd from transformers import AutoModelForCausalLM, Trainer, TrainingArguments from datasets import load_from_disk hvd.init() # Each worker loads only its shard local_rank = hvd.local_rank() world_size = hvd.size() torch.cuda.set_device(local_rank) # Load dataset shard for this worker dataset = load_from_disk(f"/dbfs/tmp/llm-finetune/train_shards/shard_{local_rank}") model = AutoModelForCausalLM.from_pretrained( BASE_MODEL, torch_dtype=torch.bfloat16, ).to(f"cuda:{local_rank}") # Wrap optimizer with Horovod DistributedOptimizer optimizer = torch.optim.AdamW(model.parameters(), lr=hparams["lr"]) optimizer = hvd.DistributedOptimizer( optimizer, named_parameters=model.named_parameters(), compression=hvd.Compression.fp16, # compress gradient communication ) # Broadcast initial model weights from rank 0 to all workers hvd.broadcast_parameters(model.state_dict(), root_rank=0) hvd.broadcast_optimizer_state(optimizer, root_rank=0) training_args = TrainingArguments( output_dir=f"/dbfs/tmp/llm-finetune/hvd_output", num_train_epochs=hparams["epochs"], per_device_train_batch_size=hparams["batch_size"], bf16=True, no_cuda=False, dataloader_num_workers=2, # Only rank 0 logs and saves — avoids duplicated artifacts report_to="mlflow" if hvd.rank() == 0 else "none", save_strategy="epoch" if hvd.rank() == 0 else "no", ) trainer = Trainer( model=model, args=training_args, train_dataset=dataset, optimizers=(optimizer, None), ) trainer.train() # Only rank 0 registers the model if hvd.rank() == 0: mlflow.transformers.log_model( transformers_model={"model": model, "tokenizer": tokenizer}, artifact_path="model", registered_model_name=MODEL_NAME, ) # Launch distributed training across N GPU workers # np = number of processes = number of GPUs across all nodes hr = HorovodRunner(np=8, driver_log_verbosity="all") # 8 GPUs (e.g., 2 × 4-GPU nodes) hr.run(train_fn, hparams={ "lr": 2e-5, "epochs": 3, "batch_size": 2, # per GPU; effective = 2 × 8 = 16 }) MLflow Model Registry and Promotion Once a run completes, models land in the MLflow Model Registry. Databricks uses Unity Catalog-backed model aliases (candidate, staging, champion) instead of the legacy stage model. Python # ── Model Registry Promotion Workflow ───────────────────────────────────────── from mlflow.tracking import MlflowClient client = MlflowClient() # Get the latest registered version from the training run latest_version = client.get_registered_model(MODEL_NAME).latest_versions[0].version # Tag the new version as a candidate for review client.set_registered_model_alias( name=MODEL_NAME, alias="candidate", version=latest_version, ) client.set_model_version_tag( name=MODEL_NAME, version=latest_version, key="fine_tuned_on", value="gold.support_conversations", ) client.set_model_version_tag( name=MODEL_NAME, version=latest_version, key="eval_loss", value=str(round(eval_results["eval_loss"], 4)), ) # After human review / automated eval gates pass → promote to staging client.set_registered_model_alias( name=MODEL_NAME, alias="staging", version=latest_version, ) # After integration tests pass → promote to champion (production) client.set_registered_model_alias( name=MODEL_NAME, alias="champion", version=latest_version, ) # Load model by alias — decouples code from version numbers champion_model = mlflow.transformers.load_model(f"models:/{MODEL_NAME}@champion") Serving With Databricks Model Serving Python # ── Deploy to Databricks Model Serving ──────────────────────────────────────── # Can also be done via the UI: Models > Serving > Create Endpoint import requests, json WORKSPACE_URL = "https://<your-workspace>.azuredatabricks.net" TOKEN = dbutils.secrets.get("prod-scope", "databricks-token") endpoint_config = { "name": "support-intent-classifier", "config": { "served_models": [ { "name": "mistral-7b-lora-champion", "model_name": MODEL_NAME, "model_version": latest_version, "workload_size": "Small", # 1 GPU "scale_to_zero_enabled": True, "workload_type": "GPU_LARGE", # A10G } ], "traffic_config": { "routes": [ {"served_model_name": "mistral-7b-lora-champion", "traffic_percentage": 100} ] }, "auto_capture_config": { "catalog_name": CATALOG, "schema_name": "ml", "table_name": "support_classifier_inference_log", "enabled": True, # log all requests/responses to Delta } } } response = requests.post( f"{WORKSPACE_URL}/api/2.0/serving-endpoints", headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}, data=json.dumps(endpoint_config), ) print(response.json()) # ── Query the endpoint ──────────────────────────────────────────────────────── def classify_intent(message: str) -> str: payload = { "inputs": {"prompt": f"[INST] Classify the intent of this support message:\n\n{message} [/INST]"}, "params": {"max_new_tokens": 50, "temperature": 0.1}, } resp = requests.post( f"{WORKSPACE_URL}/serving-endpoints/support-intent-classifier/invocations", headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}, data=json.dumps(payload), ) return resp.json()["predictions"][0] print(classify_intent("My order hasn't arrived and it's been 10 days")) # → "shipping_delay" Comparing Fine-Tuning Strategies StrategyGPU MemoryTraining TimeQuality vs Full FTWhen to UseFull Fine-TuningVery High (80GB+)SlowestBaseline (100%)Max quality, large budgetLoRAMedium (24–40GB)Fast~95%Best general-purpose choiceQLoRA (4-bit + LoRA)Low (10–16GB)Medium~90–93%Single GPU, cost-sensitivePrefix TuningLowVery Fast~80–85%Minimal compute, quick iterationPrompt TuningVery LowFastest~70–80%Inference-only, no weight changeRLHF / DPOHighSlowestBest alignmentInstruction-following qualityDistillationMedium (teacher)MediumVariesSmaller, faster inference model Rule of thumb: Start with QLoRA on a single GPU. If eval loss stagnates or quality gates fail, move to LoRA on multi-GPU. Full fine-tuning is only warranted when you have >1M high-quality labeled examples and a measurable business case for the incremental quality gain. Key Takeaways Spark handles data at scale before training even begins — filtering, tokenization, and splitting across millions of records in minutes.QLoRA + LoRA makes fine-tuning 7B–13B models accessible on a single A100, reducing memory footprint by ~70% with minimal quality loss.MLflow report_to="mlflow" gives you automatic experiment tracking with zero extra code — every loss curve, gradient norm, and learning rate schedule is captured.Unity Catalog model aliases (candidate → staging → champion) replace brittle version-number references in deployment code, making promotions and rollbacks a one-liner.Auto Capture on Databricks Model Serving logs every inference request and response to a Delta table — giving you a feedback loop to build your next training dataset.Horovod on Spark is the right tool when single-node training exceeds your SLA — it leverages your existing Spark cluster without a separate orchestration layer. References Databricks — LLM Fine-Tuning on DatabricksMLflow — Transformers Flavor DocumentationHugging Face PEFT — LoRA & QLoRAQLoRA Paper — "QLoRA: Efficient Finetuning of Quantized LLMs" (Dettmers et al., 2023)LoRA Paper — "LoRA: Low-Rank Adaptation of Large Language Models" (Hu et al., 2021)Databricks — Model Serving (Foundation Model APIs)Horovod on Spark — Official DocumentationDatabricks — HorovodRunner APIDatabricks — Inference Tables (Auto Capture)"Training language models to follow instructions with human feedback" — InstructGPT / RLHF (OpenAI, 2022)

By Jubin Abhishek Soni CORE

A multi-SLM platform creates value only when specialization does not introduce a new latency tier. Small language models are inexpensive enough to dedicate to focused work such as extraction, code handling, safety filtering, or short-form reasoning, but that advantage disappears if model selection itself becomes expensive. Research on LLM routing shows that query difficulty varies enough for model choice to materially affect efficiency and quality, and modern serving stacks expose enough control over routing, batching, and cache locality to turn that insight into an operational design rather than an academic one. In practice, the routing layer has to behave like a tiny data-plane decision engine, not like another inference hop. Why Multiple SLMs Need Routing A single small model rarely gives the best latency-quality trade-off for every prompt type. Short structured requests, such as JSON extraction and classification, differ sharply from code repair, and both differ again from prompts that need broader reasoning. RouteLLM describes routing as assigning simpler queries to weaker models and reserving stronger models for harder cases, while FrugalGPT reports that a learned cascade can preserve strong-model quality with very large cost reductions. Although those papers evaluate broader LLM portfolios, the underlying lesson transfers cleanly to a fleet of small specialized models: heterogeneity in request shape makes heterogeneity in model choice economically and operationally rational. That conclusion rules out a router that behaves like another generative model call. RouteLLM explicitly treats effective routing as a pre-decision that minimizes cost and latency relative to broader multi-model execution, which means the dominant path should remain inside in-memory feature extraction and lookup. Prompt length, requested output shape, language, code markers, safety category, session identity, and prior cache affinity are all signals that can be computed before any model is invoked. A practical design target is to keep that first decision under a millisecond, so its cost remains far below prefill and decode work. The moment the main path depends on an additional model inference, the latency budget starts competing with the very SLM call it is supposed to optimize. Keeping the Decision Path Short The cleanest design is a two-stage router. The first stage is deterministic and resolves obvious cases immediately. A short request demanding strict JSON can go to an extraction model. A prompt containing fenced code, compiler errors, or repository paths can go to a code model. A safety-sensitive request can be pinned to a policy model. Only when simple predicates fail to produce a confident mapping should the second stage run, and that second stage should be a lightweight complexity scorer rather than another generator. Ray Serve’s request-routing API is built around this kind of custom replica selection, and its FIFO mixin is specifically intended for algorithms that can route requests as soon as they arrive without waiting for content-heavy processing. That is the right shape for an ultra-low-latency router: deterministic fast path first, optional scorer second. A routing metadata object makes that design practical because it compresses request interpretation into cheap primitives: Java record RoutingContext( int tokenCount, boolean codeRequest, boolean structuredOutput, String language, boolean repeatedPrefix, double complexityScore ) {} This record is deliberately plain. Primitive fields are cheap to serialize, cheap to log, and easy to replay during debugging. That choice aligns with PyTorch and vLLM production notes on disaggregated serving, where complex metadata objects in scheduler paths increased serialization cost and hurt inter-token behavior, and it fits the general shape of request routers that repeatedly rank candidate replicas under load. The complexityScore field should therefore come from a compact classifier or calibrated heuristic trained offline on task outcomes, escalation rates, or preference labels, not from a runtime SLM call. The router’s intelligence belongs in the thresholds and features, not in an extra generation step. The routing function should then read like admission control rather than orchestration: Java ModelTarget route(RoutingContext ctx) { if (ctx.structuredOutput() && ctx.tokenCount() < 800) return ModelTarget.EXTRACTION_SLM; if (ctx.codeRequest()) return ModelTarget.CODE_SLM; if (ctx.complexityScore() > 0.72) return ModelTarget.REASONING_SLM; if (ctx.repeatedPrefix()) return ModelTarget.GENERAL_SLM_CACHE_HOT; return ModelTarget.GENERAL_SLM; } The important detail is ordering. The cheapest predicates run first, the optional scorer appears only after clear task signals have been checked, and cache affinity refines the generic path instead of overriding obvious specialization. That mirrors how high-performance request routers rank candidates and then filter out replicas that are already saturated. Thresholds should be calibrated from observed latency and task-success data, but the architectural rule is stable: most traffic should leave the router with a decision produced entirely from fields already in memory. Making Selection Cache-Aware Cache-aware selection is where routing often starts to produce visible latency gains. vLLM’s automatic prefix caching reuses KV cache from earlier queries when a new request shares the same prefix, allowing shared prompt computation to be skipped, and its design notes describe prefix caching as close to a free lunch because it avoids redundant work without changing outputs. SGLang reaches a similar result with RadixAttention, which keeps reusable KV state in a radix tree, adds LRU eviction, and applies cache-aware scheduling to improve hit rate while introducing only negligible overhead when no cache hit occurs. That combination matters because a fast model on a warm prefix can easily outperform a nominally better model on a cold path. Routing without cache awareness, therefore, leaves substantial latency savings on the table. That is why a field such as repeatedPrefix, promptFamilyId, or session hash belongs in the routing context. Ray Serve exposes locality-aware and multiplex-aware helpers so that requests can prefer nearby replicas or replicas that already hold the relevant model, and Meta’s PyTorch and vLLM production write-up reports that sticky routing of the same session to the same prefill host significantly boosts prefix-cache hit rate, reaching 40% to 50% hit rate in the described deployment. The practical lesson is broader than that specific architecture. Similar prompt families should be steered toward the same warm replicas whenever possible, even if a purely load-balanced policy would have spread them evenly. Equal distribution is not the same thing as minimal latency once KV reuse becomes available. Keeping the System Fast in Production Once the routing logic is correct, the queueing policy and replica shape become the next sources of latency. Triton documents that dynamic batching combines requests to maximize throughput and allows bounded queue delay, while concurrent model execution and instance groups allow multiple copies of the same model to run in parallel on selected devices. That argues for selective rather than universal batching. Short extraction or moderation SLMs often benefit from aggressive batching because their service time is small and predictable, while interactive reasoning models need tighter queue-delay bounds to prevent batching from inflating p95 latency. Replica placement matters as well. Heavy or frequently chosen models deserve more parallel instances, and cold-start penalties should be reduced through explicit warmup, since Triton notes that model warmup can prevent the slow initial inferences seen before a model is fully initialized. Backpressure and observability complete the design. Ray Serve supports bounded queues and load shedding through max_queued_requests, and its autoscaling guidance ties lower ongoing-request targets to tighter latency objectives. Ray Serve LLM also exposes request latency, throughput, TTFT, and TPOT, while Triton exposes Prometheus metrics for GPU and request behavior. Those signals should be segmented by routed model, decision path, cache-hit class, and warm versus cold replica so that routing regressions become visible before they surface as user-facing tail latency. Without route-level telemetry, an apparently accurate router can quietly push traffic onto cold replicas, oversized queues, or cache-miss-heavy paths. In a low-latency SLM system, observability is not just for debugging. It is the only reliable way to keep routing policy aligned with actual serving behavior. Conclusion An ultra-low-latency routing layer for multiple SLMs is best treated as a serving primitive rather than as a separate intelligence feature. The strongest design keeps most requests on a deterministic first stage, invokes a lightweight complexity scorer only for ambiguous prompts, represents route state with compact metadata, and treats prefix locality as a first-class selection signal. Around that core, warm replicas, selective batching, bounded queues, and route-level observability determine whether specialization actually improves latency or merely rearranges it. When routing is cheaper than a single token step and cache locality is preserved instead of ignored, a multi-SLM system stops looking like a collection of models and starts behaving like a disciplined low-latency inference fabric.