Data

Every on-call engineer understands this situation well. An alert fires at 2 a.m., engineers spend the first five minutes figuring out what it means, the next few minutes searching Confluence for the relevant runbook, and finally start doing something useful. By that point, an automated system that could have classified the alert and retrieved the right procedure, proposed a remediation plan, and opened a ticket in thirty seconds has saved you nothing because it didn’t exist. That’s the problem this article addresses. We are going to build a working incident triage agent using .NET 10 and .NET Aspire 9 that does exactly that chain of steps automatically. The agent receives an HTTP alert payload, which classifies it using a Groq-hosted LLM, retrieves the matching runbook section from a Qdrant vector store, asks the LLM to propose remediation steps, and escalates to PagerDuty (through a local stub). If the severity warrants it, it writes a full audit record. The system will automatically follow all these without human involvement. What makes this integration not have any single component? It’s the combination of MCP as the tool contract, Aspire as the wiring layer, and a small eval harness that prevents the agent from quietly drifting over time. Let's go through how it's built. What are we Actually Solving? Before we bring AI into this, let’s be honest about what the actual problem is because “AI for incident triage” sounds impressive but means nothing without a clear picture of what exactly the AI is doing. When an alert goes out, the on-call engineer has three jobs Is this serious? (figuring out the severity before doing anything else)What do I do about it? (find the right procedure and follow it)Who else needs to know? (escalate the right people and open a ticket) Here are the things: Job one is mostly pattern matching, Job two is where it gets interesting, and Job three is completely mechanical but needs to be well understood, including failure modes, database connection pool exhaustion, memory leaks, and disk pressure. So, the answer is already written down somewhere in your runbook. Because the engineer is not thinking; they are searching. An LLM is genuinely good at steps one and two when given the right context. It can classify alerts and turn a runbook excerpt into a clear list of actions. The tricky part is making sure it gets the right runbook excerpt in the first place. If you ask an LLM to fix a memory leak without giving it the memory runbook, you’ll get a generic answer. So, give the right context, and you get something genuinely useful. Solution Architecture The solution is split into six focused .NET Aspire projects. Each project has a single, well-defined responsibility. App Host is the entry point to run. It doesn’t serve HTTP traffic or business logic. Its only job is to tell Aspire what services exist, which one needs to start, and which configuration needs to be injected into each of them. Think of it as the framework that describes the whole system. Services Defaults is a shared library that all other projects reference. It sets up the things every service should have, including structured logging with Serilog, distributed tracing, health check endpoints, and service discovery. So, these are all wired up with a single builder.AddServiceDefaults() call and never think about it again. Agent Service is the front door. It exposes one endpoint POST/triage and drives the five-step pipeline from start to finish. It doesn’t classify alerts, talk to Qdrant, and doesn’t know what pager duty is. It just calls the right tools in the right order and assembles the final response. MCP Tool Server is where the actual work happens. It hosts four MCP tools (alert classification, runbook lookup, PagerDuty escalation, and audit writing) and exposes them over HTTP using the Model Context Protocol. The Agent Service calls these tools by name without knowing anything about their internal implementation. PagerDuty Stub is a throwaway stand-in for the real PagerDuty API. In development, you do not want to fire real pagers or need a PagerDuty account just to test the escalation step. The stub accepts the same payload, logs it, and returns a synthetic ticket. Swap it for the real endpoints in production by changing one config value. Evals Harness is a safety check. It fires six carefully chosen alerts at the live agent and checks that the responses match expectations. If fewer than five pass, the process exits with a non-zero code, and your continuous integration pipeline fails. It is the thing that tells you when a model update or a config change has quietly broken something. The data flow for a single alert looks like this. The AgentService and McpToolServer are deliberately separate processes. The agent knows nothing about embeddings, Qdrant, or PagerDuty. It only knows how to call MCP tools by name. This is the core benefit of MCP. In the future, if we update the MCP server, the agent doesn’t change at all. MCP Tool Server The McpToolServer is an ASP.NET Core minimal API that exposes four tools over the MCP streamable HTTP transport. Each tool is a static class annotated with `McpServerToolType` and `McpServerTool`. C# [McpServerToolType] public static class AlertClassifierTool { [McpServerTool, Description("Classify an alert and return severity, category, and confidence.")] public static async Task<AlertClassification> ClassifyAsync( [Description("The raw alert text to classify")] string alertText, IChatClient chatClient, ILogger<AlertClassifierTool> logger, CancellationToken ct) { var prompt = $""" You are an incident classifier. Classify the following alert: {alertText} Respond with JSON only: {{ "severity": "Critical|High|Medium|Low", "category": "short category label", "confidence": 0.0-1.0, "reasoning": "one sentence" } """; var response = await chatClient.GetResponseAsync(prompt, new ChatOptions { ResponseFormat = ChatResponseFormat.Json }, ct); return JsonSerializer.Deserialize<AlertClassification>(response.Text) ?? throw new InvalidOperationException("LLM returned empty classification"); } } The `IChatClient` and `ILogger` parameters are injected by the MCP framework via ASP.NET Core’s dependency injection container. The tool itself is stateless, a plain static method. This keeps unit testing straightforward and allows you to pass in a mock `IChatClient`, call the method, and assert on the result. The `RunbookLookupTool` follows the same pattern but takes an `IEmbeddingGenerator<string, Embedding<float>>` and a `QdrantClient` instead of a chat client. C# [McpServerTool, Description("Find the most relevant runbook excerpts for a given incident category.")] public static async Task<List<RunbookExcerpt>> LookupAsync( [Description("Incident category from classification")] string category, IEmbeddingGenerator<string, Embedding<float>> embedder, QdrantClient qdrant, IConfiguration config, CancellationToken ct) { var topK = int.Parse(config["Qdrant:TopK"] ?? "3"); var colName = config["Qdrant:CollectionName"] ?? "runbooks"; var embedResult = await embedder.GenerateAsync([category], cancellationToken: ct); var vector = embedResult[0].Vector.ToArray(); var hits = await qdrant.SearchAsync(colName, vector, limit: (ulong)topK, cancellationToken: ct); return hits.Select(h => new RunbookExcerpt( Title: h.Payload["title"].StringValue, Content: h.Payload["content"].StringValue, Score: (float)h.Score)).ToList(); } The vector query uses cosine similarity, so (high memory usage on API node) still finds the memory-pressure runbook even though the wording doesn’t match. The embeddings capture semantic meaning, not keyword overlap. Custom Embeddings with Nomic AI Nomic AI `nomic-embed-text-v1.5` model produces 768-dimensional vectors at very low cost. The only catch is that Nomic uses a non-standard API path (`POST/v1/embedding/text` rather than the OpenAI-compatible `/V1/embeddings`), so we can’t use the default OpenAI embedding adapter from `Microsoft.Extensions.AI`. Instead, we implement `IEmbeddingGenerator<string, Embedding<float>>` directly. C# internal sealed class NomicEmbeddingGenerator( IHttpClientFactory httpClientFactory, string model, ILogger<NomicEmbeddingGenerator> logger) : IEmbeddingGenerator<string, Embedding<float>> { public EmbeddingGeneratorMetadata Metadata { get; } = new("nomic", providerUri: null, defaultModelId: model); public async Task<GeneratedEmbeddings<Embedding<float>>> GenerateAsync( IEnumerable<string> values, EmbeddingGenerationOptions? options = null, CancellationToken cancellationToken = default) { var client = httpClientFactory.CreateClient("nomic"); var requestBody = new NomicEmbedRequest(model, values.ToList(), "search_document"); using var response = await client.PostAsJsonAsync( "embedding/text", requestBody, NomicJsonContext.Default.NomicEmbedRequest, cancellationToken); response.EnsureSuccessStatusCode(); var result = await response.Content.ReadFromJsonAsync( NomicJsonContext.Default.NomicEmbedResponse, cancellationToken) ?? throw new InvalidOperationException("Nomic returned an empty response body"); return new GeneratedEmbeddings<Embedding<float>>( result.Embeddings.Select(v => new Embedding<float>(v)).ToList()); } public object? GetService(Type serviceType, object? serviceKey = null) => null; public void Dispose() { } } This class implements the full `IEmbeddingGenerator<string, Embedding<float>>` contract from `Microsoft.Extensions.AI`, so the rest of the codebase, including the `RunbookLookupTool`, sees a standard interface and never needs to know it’s talking to Nomic rather than OpenAI. The `JsonSerializable` source generation at the bottom of the file `NomicJsonContext` is important for trimming-safe serialization and for performance in hot paths. Both the request and response records must be at namespace scope (not nested inside the generator class) for the source generator to work correctly. This is a common mistake that produces `SYSLIB1032` at compile time. The Agent Service The Agent Service is where the triage pipeline is assembled. It uses Semantic Kernel to handle the remediation step (where we need prompt rendering and the injection filter) and calls all other steps via `McpClient.CallToolAsync`. The pipeline in `DotNetAspireTriageAgentService.cs` looks like this. C# // Step 1 — Classify var classification = await _mcpClient.CallToolAsync<AlertClassification>( "ClassifyAsync", new { alertText = payload.AlertText }, ct); // Step 2 — Runbook lookup (skip for Medium/Low) List<RunbookExcerpt> runbooks = []; if (_lookupSeverities.Contains(classification.Severity)) { runbooks = await _mcpClient.CallToolAsync<List<RunbookExcerpt>>( "LookupAsync", new { category = classification.Category }, ct); } // Step 3 — Remediation (via Semantic Kernel for prompt filter support) var proposal = await _kernel.InvokePromptAsync<RemediationProposal>( RemediationPromptTemplate, new KernelArguments { ["alert"] = payload.AlertText, ["runbooks"] = JsonSerializer.Serialize(runbooks), ["severity"] = classification.Severity }, cancellationToken: ct); // Step 4 — Escalate var escalation = await _mcpClient.CallToolAsync<EscalationResult>( "EscalateAsync", new { classification, correlationId = payload.CorrelationId }, ct); // Step 5 — Audit await _mcpClient.CallToolAsync( "WriteAuditAsync", new { classification, proposal, escalation }, ct); Defending Against Prompt Injection Prompt injection is a real concern in agentic systems where user-supplied text ends up literally inside an LLM prompt. An attacker who controls the alert body could try to override the system prompt and redirect the agent’s behavior. Prevent here uses Semantic Kernel’s `IPromptRenderFilter`, which fires after the prompt template is rendered but before the rendered string is sent to the model. C# public sealed class PromptInjectionFilter( InjectionDetectionContext context, ILogger<PromptInjectionFilter> logger) : IPromptRenderFilter { // Matches common injection patterns: "ignore previous instructions", // "disregard your system prompt", role-switching attempts, etc. private static readonly Regex InjectionPattern = new( @"(?i)(ignore\s+(all\s+)?(previous|prior|above)\s+instructions?" + @"|disregard\s+(your\s+)?(system\s+prompt|instructions?)" + @"|you\s+are\s+now\s+(?:a\s+)?(?:an?\s+)?\w+" + @"|act\s+as\s+(if\s+you\s+are\s+)?(?:a\s+)?(?:an?\s+)?\w+)", RegexOptions.Compiled | RegexOptions.CultureInvariant); public async Task OnPromptRenderAsync( PromptRenderContext context, Func<PromptRenderContext, Task> next) { await next(context); // let the template render first if (context.RenderedPrompt is not null && InjectionPattern.IsMatch(context.RenderedPrompt)) { context.RenderedPrompt = InjectionPattern.Replace( context.RenderedPrompt, "[SANITISED]"); this.context.InjectionDetected = true; logger.LogWarning( "Prompt injection attempt detected and sanitised — correlationId={CorrelationId}", context.Arguments["correlationId"]); } } } The filter doesn’t abort the request. It sanitizes the offending text and sets a flag that the agent includes in the response. This is a deliberate choice where failing silently is worse than completing with a sanitized prompt, because a failed triage means a missed escalation. The response `injectionDetected` field lets downstream systems know that something suspicious happened without stopping the pipeline. Handle Everything Together with .NET Aspire The AppHost is where everything comes together. Every service, dependency, and API key is declared in one place. When we run this project, Aspire reads those declarations and automatically starts the entire system in the correct order. C# var builder = DistributedApplication.CreateBuilder(args); // API keys from user-secrets or appsettings.json var groqApiKey = builder.AddParameter("GroqApiKey", secret: true); var nomicApiKey = builder.AddParameter("NomicApiKey", secret: true); // Qdrant container — persisted between restarts var qdrant = builder.AddQdrant("vectorstore") .WithLifetime(ContainerLifetime.Persistent); // PagerDuty development stub var pagerDutyStub = builder.AddProject<Projects.DotNetAspireTriageAgent_PagerDutyStub>( "pagerduty-stub"); // MCP Tool Server — waits for Qdrant and the PagerDuty stub var pagerDutyStubEndpoint = pagerDutyStub.GetEndpoint("http"); var mcpServer = builder.AddProject<Projects.DotNetAspireTriageAgent_McpToolServer>("mcp-tools") .WithReference(qdrant) .WithReference(pagerDutyStub) .WaitFor(qdrant) .WaitFor(pagerDutyStub) .WithEnvironment("Groq__ApiKey", groqApiKey) .WithEnvironment("Nomic__ApiKey", nomicApiKey) .WithEnvironment("PagerDuty__StubEndpoint", ReferenceExpression.Create($"{pagerDutyStubEndpoint}/pagerduty-stub/incidents")); // Agent Service — waits for the MCP server builder.AddProject<Projects.DotNetAspireTriageAgent_AgentService>("agent-service") .WithReference(mcpServer) .WaitFor(mcpServer) .WithEnvironment("Groq__ApiKey", groqApiKey); builder.Build().Run(); Three things in this code are worth understanding properly before moving on. .WithReference() vs .WithEnvironment(): These two look similar but do various jobs. When you call .WithReference(Qdrant), you are telling Aspire to figure out Qdrant’s host, port, and credentials at runtime and automatically inject the full connection string into McpToolServer. We do not need to mention it hardcoded anywhere. ReferenceExpression.Create. This one trips people up the first time. When McpToolServer needs to call the PagerDuty stub, it needs the stub’s full URL including the path (like domain/pagerduty-stub/incidents). The problem is you do not know the port number at the time you write the code; in this case, Aspire assigns it dynamically at startup. So instead of hardcoding a URL that will break on someone else’s machine, for this we write ReferenceExpression.Create($"{pagerDutyStubEndpoint}/pagerduty-stub/incidents") and let Aspire fill in the real address when it starts up. WaitFor This tells Aspire not to start McpToolServer until Qdrant and the PagerDuty stub are fully up and ready. Without it, McpToolServer would try to connect before they are ready and crash on the very first run. Once everything is running, the Aspire dashboard gives you a live view of the whole system. The resources tab shows all four services with their current health status and the URLs Aspire assigned to each one. The graph tab is even more useful when you are onboarding someone new to the project. It draws the exact dependency map you declared in the codebase, which service depends on which, which API keys go where, and how everything connects. Note: if a service fails to start, this graph tells you immediately which dependency in the chain is the problem instead of you having to read through logs across four different console windows. PagerDuty Stub Rather than mocking PagerDuty calls in covers or requiring a real PagerDuty account, the solution includes a lightweight stub service. It is a genuine Aspire project registered in Apphost. C# app.MapPost("/pagerduty-stub/incidents", async (HttpRequest request) => { // ... read and log the body ... var response = new PagerDutyStubResponse( Incident: new StubIncident( Id: correlationId, Status: "triggered", Number: Random.Shared.Next(1000, 9999))); return Results.Created( $"/pagerduty-stub/incidents/{correlationId}", response); }); Because the stub is a real Aspire project, its URL is dynamically allocated by Aspire and injected into McpToolServer via `ReferenceExpression.Create`. This means there are no hardcoded ports that break when someone else is already using that port, and the stub starts and stops with the rest of the solution. Swapping it for the real PagerDuty events API in Production means changing a single config value, the URL injected via `WithEnvironment`. Runbook Seeding on Startup The McpToolServer seeds its Qdrant collection on startup using a hosted service. It checks whether the collection already exists before doing any work, which means subsequent restarts are near-instant. C# public sealed class RunbookSeeder( QdrantClient qdrant, IEmbeddingGenerator<string, Embedding<float>> embedder, IConfiguration config, ILogger<RunbookSeeder> logger) : IHostedService { public async Task StartAsync(CancellationToken ct) { var collectionName = config["Qdrant:CollectionName"] ?? "runbooks"; var exists = await qdrant.CollectionExistsAsync(collectionName, ct); if (exists) { logger.LogInformation("Runbook collection already exists — skipping seed"); return; } await qdrant.CreateCollectionAsync(collectionName, new VectorsConfig(new VectorParams(size: 768, distance: Distance.Cosine)), ct); } } The runbooks test the most common failure categories, including high CPU, memory pressure, database connection exhaustion, disk saturation, network timeout, and pod restart loops. Each is stored as a Qdrant point with title and content payload fields that `RunbookLookupTool` reads back on retrieval. Eval Harness AI systems have a subtle problem that unit tests don’t catch. So, the agent can quietly get worse over time. A model version bumps, someone tweaks a prompt, a config value changes, and suddenly your critical alerts are coming back as medium with no error thrown anywhere. You only find out when a real incident gets missed. The Evals project is the safety net for exactly this. It fires six alert payloads at the live agent and checks that each response matches the expected severity, category, and escalation behavior. If fewer than five pass, the build fails. It is the same idea as a unit test suite, except it is testing the intelligence of the agent, not just the correctness of the code. Key Takeaways .NET Aspire service coordination makes it practical to run a multi-service AI agent system, including a vector database, an MCPToolServer, and an LLM-backed agent, locally with a single `dotnet run` command.The Model Context Protocol (MCP) gives you clean, language-agnostic control for exposing agent tools over HTTP, so the agent and its capabilities can evolve independently without tight coupling.Combining Nomic AI embeddings with a Qdrant vector store lets you attach a runbook knowledge base to an AI agent without fine-tuning a model that will help semantic search retrieve the right production even when the alert wording doesn’t match the runbook text exactly.Groq’s OpenAI-compatible API with `llama-3.3-70v-versatile` provides sub-second structured JSON responses, which is fast enough to complete a full five-step triage pipeline including classify, retrieve, remediate, escalate, and audit in under three seconds on most workloads.Adding a Semantic Kernel `IPromptRenderfilter` to scan every prompt render before it reaches the LLM is a lightweight, zero-overhead way to defend against prompt injection in agentic pipelines. Prerequisites To follow along with the code in this article, you will need: Visual Studio 2026 (17 or later) with the .NET Aspire package installed, or the .NET 10 SDK (10.0.300 or later) if you prefer using a terminal.Docker Desktop (4.x or later) must be running before you start because .NET Aspire automatically starts a Qdrant container.Groq API key (free get from console.groq.com) used for the alert classification and remediation via `llama-3.3-70v-versatile`.Nomic AI API key (free get from atlas.nomic.ai) used for runbook text embeddings via `nomic-embed-text-v1.5`. Note: No cloud subscription is required. Both API keys have generous free quotas that comfortably cover development and testing. Conclusion What we have built is a working blueprint for an AI triage agent that respects software engineering discipline, clean boundaries between components, a tool contract that survives dependency changes, prevents misuse, and a regression harness that makes model-level drift a continuous integration failure rather than a surprise. The combination of .NET Aspires coordination, Mcp tool abstraction, Groq’s low-latency inference, and Nomic embeddings means you can stand up a full agentic pipeline locally, with realistic dependencies, in the time it takes to run `dotnet run`. The development experience matters because it determines how quickly you can experiment, iterate, and validate changes. The next natural extensions are a persistent audit store, a document ingestion pipeline for runbooks, and a feedback loop that uses closed incidents to refine the classification prompts. All three can be added as new MCP tools without changing the agent. Appendix The complete source code for this article, including all six projects, runbook seed data, eval harness cases, and configuration examples, is available in the GitHub repository. You can clone it, run it locally with a single command, and use it as a starting point for your own incident triage pipeline. Full source code is available at the GitHub Repository.

The Problem: Our p99 Was 3-5 Seconds Our PyFlink pipeline was missing its latency SLO by seconds. The pipeline itself was straightforward: consume events from Kafka, transform them, serialize them as Protobuf, and write the results to downstream systems. Yet under production load, p99 end-to-end latency was consistently in the 3-5 second range. Profiling pointed us to an unexpected bottleneck: we were deserializing Protobuf messages in Python, even though the Flink runtime processing our stream was JVM-based. Every record that entered the Python path had to cross the JVM-to-Python process boundary, get parsed by a Python UDF, and then cross back. The business logic wasn't the problem. The doorway was. We moved Protobuf deserialization to Flink's JVM-side Protobuf format and kept Python for orchestration and SQL. In our environment, p99 dropped to approximately 500 milliseconds, with less code and a pipeline that is easier to reason about. Verified on AWS Managed Service for Apache Flink (formerly Kinesis Data Analytics). Why Python-Side Deserialization Is So Expensive The naive PyFlink architecture looks like this: A Kafka source table declared with a generic format (raw, json, or a SimpleStringSchema), so every record arrives as opaque bytes or a string.A Python map() or UDF that imports generated _pb2.py classes and calls ParseFromString() on every message.Downstream transforms and sinks. Two costs hide in step 2, and they compound at high throughput. The process boundary. PyFlink is not Python running inside Flink; it is a JVM runtime coordinating with a separate Python execution environment. Every record that enters the Python execution path incurs overhead associated with moving data between the JVM and Python, and depending on the operator and execution mode, that can involve serialization and inter-process communication in both directions. For a per-record deserialization UDF on a latency-sensitive pipeline, that overhead is paid before the actual business transformation begins. Per-record parse cost. Even when Python's Protobuf implementation uses its native backend, parsing in a Python UDF still requires the record to enter the Python execution path. When the workload is latency-sensitive and high-throughput, the combination of serialization, inter-process communication, Python execution, and parsing overhead can become significant. In our case, profiling showed that this path was a major contributor to our latency. In our pipeline, these two costs together accounted for the bulk of the gap between a 3–5 second p99 and the ~500ms target we needed, before the enrichment logic even began executing. The Key Realization: PyFlink Already Runs on the JVM Here's the insight that changes the architecture: if Protobuf is declared at the table DDL level, Flink's Kafka connector deserializes it with its native, optimized JVM-based Protobuf format before any data reaches the Python side. Your columns simply arrive typed and ready. Python's role shrinks to what it's genuinely good at in this stack: orchestration and SQL. No rewrite to Java. No change to how jobs are deployed. Just a different declaration of intent. The trade is that Flink's native Protobuf format needs compiled Java message classes on the classpath; it does not consume .proto files or Python _pb2 modules directly. That means adding a small build step to your workflow, which we'll cover below. Implementation The pipeline splits into two declarative jobs. Job 1: JSON In, Protobuf Out The source table reads the raw JSON topic; the sink table declares format = 'protobuf' and points at the compiled Java class. The JVM handles typed-row-to-Protobuf encoding. SQL -- SOURCE: raw JSON payload as STRING plus Kafka record timestamp CREATE TABLE source_events_json ( event_data STRING, kafka_timestamp TIMESTAMP(3) METADATA FROM 'timestamp' ) WITH ( 'connector' = 'kafka', 'topic' = '${INPUT_JSON_TOPIC}', 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', 'scan.startup.mode' = 'latest-offset', 'format' = 'raw' ); -- SINK: Protobuf out to Kafka (JVM handles typed row to Protobuf) CREATE TABLE sink_events_pb ( id STRING, organization_id STRING, event_ts ROW<`seconds` BIGINT, `nanos` INT>, is_active BOOLEAN, event_type STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'acme.events.pb.v1', 'properties.bootstrap.servers' = 'kafka:9092', 'format' = 'protobuf', 'protobuf.message-class-name' = 'com.acme.events.v1.EventOuterClass$EnrichedEvent' ); -- TRANSFORM: pure SQL, no Python UDFs INSERT INTO sink_events_pb SELECT JSON_VALUE(event_data, '$.id') AS id, JSON_VALUE(event_data, '$.organization_id') AS organization_id, ROW( UNIX_TIMESTAMP(NULLIF(JSON_VALUE(event_data, '$.after.event_ts'), '')), CAST(EXTRACT(NANOSECOND FROM CAST(NULLIF(JSON_VALUE(event_data, '$.after.event_ts'), '') AS TIMESTAMP_LTZ(9))) AS INT) ) AS event_ts, CAST(JSON_VALUE(event_data, '$.is_active') AS BOOLEAN) AS is_active, JSON_VALUE(event_data, '$.event_type') AS event_type FROM source_events_json; Note what's absent: no ParseFromString(), no _pb2.py imports, no Python deserialization loop. The Python program registers DDL and runs SQL. Job 2: Protobuf In, OpenSearch Out Downstream, the sanitized Protobuf topic becomes a typed source, using the same protobuf.message-class-name property, plus ignore-parse-errors so a malformed record can't poison the pipeline. SQL -- SOURCE: Protobuf from the sanitized Kafka topic CREATE TABLE kafka_source_pb ( id STRING, organization_id STRING, event_ts ROW<`seconds` BIGINT, `nanos` INT>, is_active BOOLEAN, event_type STRING, kafka_timestamp TIMESTAMP(3) METADATA FROM 'timestamp' ) WITH ( 'connector' = 'kafka', 'topic' = 'acme.events.pb.v1', 'properties.bootstrap.servers' = 'kafka:9092', 'scan.startup.mode' = 'latest-offset', 'format' = 'protobuf', 'protobuf.message-class-name' = 'com.acme.events.v1.EventOuterClass$EnrichedEvent', 'protobuf.ignore-parse-errors' = 'true' ); -- SINK: OpenSearch (JSON) CREATE TABLE opensearch_sink ( id STRING, organization_id STRING, event_ts TIMESTAMP_LTZ(3), is_active BOOLEAN, event_type STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'opensearch-2', 'hosts' = '${OPENSEARCH_ENDPOINT}:443', 'index' = 'acme-events-v1', 'format' = 'json' ); INSERT INTO opensearch_sink SELECT id, organization_id, TO_TIMESTAMP_LTZ(event_ts.seconds * 1000, 3), is_active, event_type FROM kafka_source_pb; The Build Step: Getting Java Classes Onto Flink's Classpath The one genuinely new piece of workflow is compiling your .proto definitions to Java and packaging them into the job's fat JAR. The essential Maven pieces: XML <dependencies> <dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java</artifactId> <version>3.25.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-protobuf</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>${flink.connector.kafka.version}</version> </dependency> <!-- plus your sink connectors, e.g. flink-connector-opensearch2 --> </dependencies> Two practices that made this maintainable for us: Version-control the generated Java sources (or generate them in CI from a single canonical .proto repo) and pull them in with build-helper-maven-plugin's add-source, rather than compiling .proto files in every consuming project. One schema source of truth, many consumers.Shade everything into one JAR with maven-shade-plugin, excluding signature files (META-INF/*.SF, *.DSA, *.RSA). On AWS Managed Flink, pass it via the job's JAR configuration; on self-managed Flink, drop it in lib/ or use --classpath. The full workflow: define the .proto, compile it to Java with protoc, package the fat JAR, put it on Flink's classpath, author the PyFlink job with the DDL above, then deploy and watch end-to-end p99. How We Measured the Improvement We measured end-to-end p99 latency as the time from a record landing on the source Kafka topic to the corresponding OpenSearch write being acknowledged MetricBeforeAfterp99 latency3-5s~500msSustained throughput~5,000 events/sec~5,000 events/secFlink parallelism128Python UDF parsingYesNoJVM/Python boundary on hot pathYesNoProtobuf decodingPythonJVM Results End-to-end p99 latency around 500 milliseconds in our environment at production load, down from a 3-5 second baseline, by eliminating per-record JVM-to-Python crossings and Python-side parsing on the hot pathLess code. The deserialization UDFs, the _pb2 imports, and their error handling all disappeared. What remains is DDL plus SQLSimpler and easier to operate. The pipeline now relies on Flink's Kafka connector and Protobuf format for serialization and parsing, with built-in parse-error handling, instead of hand-rolled Python parsing When This Optimization Won't Help Moving Protobuf decoding to the JVM won't automatically solve every latency problem. If your pipeline's critical path is dominated by sink backpressure, network latency, external API calls, state access, or checkpointing overhead rather than deserialization, changing the serialization path may have little effect on end-to-end latency. This optimization is most valuable when profiling specifically shows that Python execution and JVM/Python data movement are significant contributors to the critical path, which is why we'd recommend profiling first rather than applying this as a default change. When You Should Still Use Python UDFs This pattern is not "never write Python UDFs." It's "keep them off the per-record deserialization path." Python remains the right tool when: The transformation genuinely needs Python libraries (ML feature computation, model inference, specialized parsing that has no SQL equivalent).Throughput is modest and developer velocity matters more than the last hundred milliseconds.You're prototyping. Even then, declare the format natively from day one anyway; it costs nothing and you won't have to migrate later. If a UDF is unavoidable on a hot path, at least let the JVM do the deserialization first so the UDF receives typed columns rather than raw bytes. Gotchas Worth Knowing Before You Ship Property syntax varies by Flink version. Some versions use format = 'protobuf'; newer key/value descriptors prefer value.format = 'protobuf'. Check your version's docs.Enums: surface them as STRING if you need ergonomic SQL manipulation, or keep them numeric with a lookup table.Schema evolution: favor backward-compatible, additive changes with defaults. Because the compiled Java classes are baked into the JAR, a schema change means a rebuild and redeploy, so make that a deliberate, versioned step in CI rather than an afterthought. ignore-parse-errors is your safety net during rollout windows, but monitor the drop counter so it doesn't silently eat data.Benchmark end-to-end, not just the UDF: source lag, operator latency, and sink acknowledgments under production load patterns. Deserialization wins can be masked, or dwarfed, by sink backpressure.Security: lock down OpenSearch credentials and TLS; pin Kafka client versions compatible with your Flink release. Closing Thoughts We didn't rewrite the pipeline in Java. We removed an unnecessary per-record JVM-to-Python boundary from the hot path and let Flink's JVM-native Protobuf format do the work it was designed to do. If your PyFlink job parses Protobuf messages in Python today, check whether Flink's native format support can move that work into the JVM-side execution path. For latency-sensitive pipelines, eliminating unnecessary Python boundaries may be one of the highest-leverage optimizations to investigate, especially when profiling shows that serialization and Python execution are on the critical path.

Traditional vector RAG retrieves by embedding the question and finding semantically similar chunks, often augmented with lexical search, filtering, or reranking. This approach works when the answer is explicitly described in one or more chunks. However, it breaks down when the answer depends on relationships between facts. The question "Does my application depend on a compromised package?" illustrates this limitation. The vulnerable package may be several layers deep in the dependency tree, and no single chunk contains the answer. The answer emerges by following a chain of dependencies, but similarity search can struggle because the answer is distributed across multiple relationships rather than represented as a single semantic concept. GraphRAG addresses this issue by retrieving from a knowledge graph, where entities are connected through explicit relationships. However, GraphRAG is not a replacement for vector RAG. It is a retrieval paradigm for problems where relationships, graph structure, and provenance matter. For many applications, vector retrieval remains the fastest and most effective way to find semantically relevant information, while graph retrieval adds value when the answer depends on connected facts. But graph retrieval isn't a single technique; it requires deciding what to retrieve, how to retrieve it, and whether retrieval should happen in one pass or multiple stages. If these decisions are made correctly, GraphRAG can answer questions that vector search cannot; otherwise, it may result in a system that's slower and no more accurate than a well-designed vector RAG. This article focuses on retrieval. Graph construction is a separate issue, and schema quality affects everything. Here, we assume a well-built graph. The examples are based on a small software supply-chain graph. It's fictional, but the shape is from a real incident. In 2018, the npm package event-stream was compromised after a malicious dependency, flatmap-stream, was introduced into its dependency chain. The challenge is discovering the path shopping-app -> analytics-js -> event-stream -> flatmap-stream and connecting it to CVE-2024-1. No single text chunk contains this chain, and none of those package names indicate "compromised package." Scanners such as npm audit answer this question easily because they're built for this structure. GraphRAG can answer broader relationship questions around the same graph structure, especially when the answer requires combining multiple sources. Decision 1: Granularity Graph retrieval returns one of four units, from finest to coarsest. Node (for example, flatmap-stream and its attributes). Triplet (for example, flatmap-stream -[HAS_VULNERABILITY]-> CVE-2024-1). Path: like the red route. Paths are good for multi-hop and provenance questions. Subgraph: a connected region, like the payment component. Finer units are precise, but coarser units have more context, and more noise. Choose a unit that matches the question. Each mechanism, described in the next section, produces some units more naturally than others. Decision 2: The Six Mechanisms 1. Similarity Embed graph elements, retrieve the ones nearest the question vector. Most systems use this to find starting points. It answers "find things like this" questions on its own. Cypher CALL db.index.vector.queryNodes('pkgEmbeddings', 5, $questionVector) YIELD node, score RETURN node.name, score A simple query like "Which of our packages resemble this known-bad one?" is all that's needed. However, similarity over plain text embeddings has its limitations; it's blind to structure. So it only finds nodes that match the query and misses connected facts that don't resemble it. This method is useful for locating entry points and serving semantic lookups, but for anything that requires multiple hops, it's best to hand off to a structural mechanism. 2. Structural Traversal To get around the limitations of similarity searches, walk outward from the entry points, using techniques like neighbor expansion, breadth-first or depth-first search, and pathfinding. The granularity of what you collect depends on the approach; collecting neighbors gives you a local subgraph, while tracing routes between two entities gives you paths. Cypher // Entity-centric: what does the app build on? (a neighborhood) MATCH (:Package {name: 'shopping-app'})-[:DEPENDS_ON*1..2]->(dep:Package) RETURN DISTINCT dep // Connection question: how does the app reach vulnerable packages? MATCH p = (:Package {name: 'shopping-app'})-[:DEPENDS_ON*..5]->(bad:Package) WHERE (bad)-[:HAS_VULNERABILITY]->(:CVE) RETURN p For instance, the second query might return shopping-app -> analytics-js -> event-stream -> flatmap-stream, along with any other route to a vulnerable package. When the question is about risk, you want every possible path, not just the shortest one, because a second route is a second exposure. For large graphs, it is often more efficient to start from known vulnerable nodes and traverse backward, or constrain the search from the application side, depending on the query. Traversal is the cheapest, fastest, and easiest mechanism to explain, since you can read the route. But it has a weakness: fan-out. The practical depth limit depends on how constrained the walk is; unconstrained neighbor expansion grows rapidly, so shallow expansion is often preferred. A typed, direction-constrained path search, like the one mentioned earlier, prunes most of that growth and remains tractable deeper, which is why a four-hop dependency chase basically works in this case, but a generic four-hop expand-everything doesn't. Use traversal when the question is anchored on specific entities, and it's the best approach. 3. Graph Algorithms Two things are needed for traversal: a starting point, and a rule for edge selection. But what if you're missing one or both? There are two graph algorithms that can help. Personalized PageRank is useful when you have a starting point but no rule. It assigns a score to the mentioned entities, lets the score spread across the edges, and ranks nodes by the score they receive. Nodes that are highly reachable from the seeds through many strong paths receive higher scores. This helps find relevant nodes even if they're far away and don't share any words with the question. HippoRAG uses this for retrieval. Cypher CALL gds.pageRank.stream('supplyChain', { sourceNodes: $seedEntities, dampingFactor: 0.85 }) YIELD nodeId, score RETURN gds.util.asNode(nodeId).name AS entity, score ORDER BY score DESC LIMIT 10 For example, on our graph, if we seed analytics-js and CVE-2024-1, flatmap-stream receives a high score because it is strongly connected to both seed regions. Nothing in the query named it, though. Furthermore, directionality matters. For dependency graphs, reverse traversal or an appropriate projection is often required because vulnerabilities may be sink nodes. Community detection is especially useful for broad exploratory questions where no specific entity is known. A question like "what are the main risk areas across our dependencies?" is about the whole graph. The approach is to cluster the graph into communities, have an LLM summarize each community, and answer broad questions from those summaries. This is a key component of the Microsoft GraphRAG approach. On our graph, clustering gives us three communities: a payment-and-media stack, a web-framework stack, and the analytics subsystem with the compromised package. Cost is the main difference between the two. PageRank is expensive at query time because its scores depend on the seeds and can't be precomputed. On a large graph, you need to bound the projection to a region found by a similarity pass first. Community detection is expensive at index time because the LLM summaries are costly to compute and need to be redone when the graph changes. But queries after that are cheap. So pick PageRank when you have entities but no target. Pick communities when there's no starting entity. 4. Declarative Query With exact structural constraints in the question, it's best to query the graph directly. The model translates the question into a query language, like Neo4j's Text2Cypher. Schema grounding and validation significantly improve generated queries. Cypher // "Which volunteer-maintained packages also have a known CVE?" MATCH (pkg:Package)-[:MAINTAINED_BY]->(:Maintainer {kind: 'volunteer'}) MATCH (pkg)-[:HAS_VULNERABILITY]->(cve:CVE) RETURN pkg.name, cve.id This returns results like flatmap-stream / CVE-2024-1, combining a maintainer condition and a vulnerability condition that fuzzy mechanisms can only approximate, resulting in an exact and auditable outcome. This risk lies in the generation step, with models often inventing relationship types that sound right but don't exist in the schema; a well-formed query over imaginary edges returns zero rows without error. To mitigate this, ground the generation in the actual schema, validate the query before running it, and treat an empty result as a signal to fall back to another mechanism. Use this approach for questions that reduce to filters, counts, or joins across relationship types. 5. Generative Retrieval This mechanism works in two stages: first generating a retrieval plan that defines the relationship pattern to follow, and then translating that plan into a graph query. Reasoning on Graphs (RoG) works this way, with an LLM generating planning paths and the system retrieving the concrete paths that satisfy them. Cypher plan ← LLM("what relation path answers this?", schema) → "DEPENDS_ON* , then HAS_VULNERABILITY" paths ← graph.match(seed='shopping-app', pattern=plan) The model infers the shape of the traversal, and the graph supplies the instances. This approach fits questions where the right pattern isn't known in advance, and you don't want to hand-write a template for it. It relies on the model understanding the schema, and a one-shot plan can't correct itself unless you make the retrieval iterative. 6. Learned Retrieval A model can be trained to do the selecting, with a graph neural network scoring nodes for relevance to the question. Approaches such as G-Retriever formulate subgraph selection as an optimization problem, including variants inspired by Steiner tree formulations. The retriever is a trained component; it embeds the question, scores candidate nodes and edges, and returns the highest-value connected subgraph as evidence. This approach has demonstrated strong accuracy on hard multi-hop benchmarks, but it comes with training, serving infrastructure, and transparency costs. They're suitable for accuracy-critical question answering over a stable schema, but they're rarely the first build. Decision 3: The Paradigm You've still got to decide how many times to go to the graph. This affects both latency and accuracy. A simple approach is one retrieval, gathering everything in a single pass, which keeps latency low and is suitable for real-time answers. Iterative retrieval is another option, involving multiple passes that build on each other, useful when one pass isn't enough. It comes in two versions: fixed-rounds and adaptive. The adaptive version stops once the model has gathered sufficient information. Most production systems opt for a multi-stage approach, chaining different mechanisms together. A common pattern is using similarity retrieval to find entry points, structural traversal to expand context, and reranking to select the final evidence set. In practice, mechanisms usually combine in specific ways. Combining vector retrieval with graph retrieval is often described as HybridRAG. Letting a large language model plan the stages at query time is referred to as agentic retrieval, which is a composition pattern rather than a new mechanism. Matching Questions to Mechanisms A production system doesn't pick a mechanism per question at runtime. During the planning phase, you assess the questions, implement two or three mechanisms that cover them, and route between those in production. This table supports that assessment. The question is about…Reach forUsual granularityWhere the cost isThings semantically like X, or finding entry pointsSimilarityNode/tripletCheap, at query timeA specific entity or the routes between twoStructural traversalNode/path/subgraphCheap, at query timeMulti-hop relevance with an unknown targetPageRankRanked nodesCompute-heavy, query timeA broad theme across the whole graphCommunity detectionSubgraph + summaryExpensive, at index timeExplicit constraints: filters, counts, joinsDeclarative queryWhatever it projectsCheap, at query timeA pattern that must be inferred from the questionGenerativePath/subgraphModerate, one LLM callAccuracy-critical hard multi-hop QALearned (GNN)SubgraphExpensive, training Start with the basics. A similarity pass for entry points and structural traversal to expand. Add mechanisms as needed. PageRank or generative planning for harder questions, declarative queries for exact constraints, community summaries for thematic breadth, and learned retrieval when accuracy justifies it. Conclusion Graph retrieval involves making three key decisions. First, you need to choose the right granularity; this could be a node, triplet, path, or subgraph, depending on the answer you're looking for. The mechanism is also crucial: it's about selecting the right approach, such as similarity, traversal, graph algorithms, declarative queries, generative planning, or learned retrieval. Then there's the model: whether to use a single-pass, iterative, or multi-stage approach. By making these decisions with your system's specific questions in mind, you can design a tailored retrieval architecture rather than relying on trial and error.

If you have spent any time inside a mid-to-large organization that has embraced AI-assisted development, you've probably seen the pattern already. Teams move fast. New apps get spun up in days. Business units that used to wait months for IT now have working tools in a week. On the surface, it looks like a win. But look a little deeper, and a different picture starts to emerge. I've seen this happen firsthand: within twelve months of an organization adopting AI-assisted development, the internal app count can double, sometimes triple. And with every new app comes a fresh copy of the customer table, a slightly different definition of what a "transaction" means, and another team that has no idea what the team next door already built. The result is two compounding problems, and most organizations are treating them as if they're separate issues when they share the same root cause. The Two Problems Nobody Is Connecting There are two main problems that are impacting companies developing and deploying AI apps. They are: App Sprawl: Dozens of small applications accumulate. Each needs maintenance, security patches, dependency updates, and an owner. Most were built fast and designed by no one; they were generated. I have watched engineering teams burn entire sprints just cataloging what exists, let alone maintaining it. The long tail of unmaintained micro-apps quietly becomes an engineering liability. Data Scattering: The same business entities, customers, products, orders, and employees are defined slightly differently in every application. No canonical version exists anywhere. The same customer record lives in six places with six slightly different schemas. Reporting turns out to be like being an archaeologist! Integrations become fragile. Resuming any reconstruction means untangling a whole lot of divergent assumptions over the course of months. Most organizations look at them as individual issues: App governance is one, and data warehouse is the other. They come late and cure both the symptoms and not the cause. The actual root cause? No shared platform layer makes it structurally easy to build new applications without duplicating data and easy to share capabilities without reinventing them. Every new app starts from scratch. It creates its own database, its own auth, its own version of "what a customer is." The AI assistant helping build it has no way to know what already exists. So it builds freshness every time. The problem isn't that developers are building too much. The problem is that nothing they build connects to a common foundation. Introducing the Tectonic AI Platform The Tectonic AI Platform has been the architecture I've been working on that's actually a response to this. The governing idea is borrowed from geology: just as tectonic plates form the stable foundation beneath the dynamic surface of the earth, a Tectonic Platform provides a stable, canonical data and service layer beneath the fast-moving applications built on top of it. Applications are surface features fast to build, easy to replace, and expendable. The plate beneath them is the source of truth. It doesn't care what sits on top. It endures. This is not a product you install. It is an architectural posture, a set of structural decisions that organizations adopt before the sprawl begins or use to bring order after it already has. One important distinction worth making upfront: this is not a data warehouse. A warehouse is downstream and read-only. It doesn't stop three apps from each maintaining their own operational definition of a customer; it just lets you query all three versions in one place. The Tectonic plate is operational and live. It sits in the application layer, not below it. Apps read and write through it. It is the authoritative version, not a copy of one. The Four Pillars The framework is organized in this way. Each pillar addresses a specific failure mode that I've seen emerge when organizations skip the foundation. Pillar 1: Canonical Data Plates Shared, versioned data domains are owned by the platform, not by any single application. They include customers, products, transactions, and employees. These live on the plate. Applications interact with them through defined contracts (APIs), never by owning the underlying data store. Any app can read from the plate. Writing to it requires going through the contract. That's the word "owned by the platform" that is to be taken into account. I've seen people go to such trouble as trying to choose one app as the system of record to solve this problem. But that is no good — it would move ownership depending on how many people are on the roster. The plate does not belong to anyone; it is only legal to host the platform. Pillar 2: App Scaffolding Layer A generator framework that provisions new applications pre-wired to the plate layer from day one is also needed. When a developer or an AI assistant spins up a new app, it inherits auth, logging, observability, and data contracts automatically. The app starts connected, not isolated. Vibe coding stays fast. The structure comes for free. This is the foundation upon which the entire framework is designed to be interoperable with AI-assisted development. You aren't stopping anybody; you are just ensuring that the thing that they build into something also plugs in. Pillar 3: Capability Registry Organizations then need a discoverable catalog of everything that already exists, including APIs, workflows, AI models, reports, and integrations. Before building anything, developers (and AI coding assistants) query the registry first. Duplication becomes visible before it happens. "Does a customer lookup API already exist?" becomes a question with an answer. This is actually one of the most powerful pillars that are easy to acquire in practice. The overduplication is a mistake because people did not realize that it already existed. This is where the Register comes in. It also provides AI assistants with a surface to query before generating new code, changing the default from "build fresh" to "reuse first." Pillar 4: Governance at the Seam Rules and reviews live at the boundary between apps and plates. They are not inside individual apps. A new app can be built freely and quickly. What is allowed to be written on the plate is governed. This separates the fast surface (application layer) from the stable core (plate layer). Speed doesn't get sacrificed. Data integrity doesn't either. I want to make it clear what this pillar is NOT: it's not a committee, it's not a "ticket queue," and it's not a "review board." Governance at the seam should be automated wherever possible, including contract validation, schema versioning checks, and write permission enforcement. It's all about guardrails, not gatekeeping. What This Prevents Five Years From Now Without a Tectonic layer, here's what the organization typically looks like five years into an AI-assisted development culture: A long tail of unmaintained micro-apps, each with its own auth, its own schema, its own error handlingEngineers are spending more time stitching data together than building new capabilitiesAn AI-assisted development culture that has paradoxically made the codebase harder to understand because the surface area has exploded without any unifying structureRebuilding the same core capabilities repeatedly across teams that never knew the others existed A Tectonic layer is now in place, and every new application, no matter how quickly it is created, takes on its structure. The transformative era of vibe coding keeps on rolling. Technical "debt" is not compounded. Speed Without Structure Is Just Faster Entropy The Tectonic AI Platform is not anti-AI and not anti-speed. It is the infrastructure argument for why AI-assisted development can scale inside an organization without eventually collapsing under its own weight. The organizations that define their plates early, their canonical data domains, their shared capability contracts, and their scaffolding standards will find in a few years that they have a large and growing estate of AI-generated applications that actually work together. Those who don't will have a different, large, and growing estate. And a much harder problem to fix. The plate layer is what makes the speed sustainable. Define it early, or spend years paying for not having done so.

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

Arrays in Java are fundamental data structures used to store elements of the same type sequentially in memory. They provide a convenient way to manage collections of data where each element is accessed by its index. Basics of Arrays An array in Java is a fixed-size container that holds a specific number of elements of the same data type. This means all elements in an array must be of the same type such as integers (int), floating-point numbers (double), characters (char) or objects (Object). Declaring and Initializing Arrays To declare an array in Java we specify the type of elements followed by square brackets [] and the array name: Java dataType[] arrayName; For example, to declare an integer array named numbers: Java int[] numbers; Arrays in Java are objects and like all objects they must be instantiated with the new keyword before they can be used: Java arrayName = new dataType[arraySize]; For instance, to create an integer array numbers with a size of 5: Java int[] numbers = new int[5]; This initializes an array numbers that can hold 5 integers with indices ranging from 0 to 4. Accessing Elements in Arrays Array elements are accessed using their index, which starts at 0 for the first element and goes up to arraySize - 1 for the last element. For example, to access and modify elements of the numbers array: Java int[] numbers = {10, 20, 30, 40, 50}; int firstElement = numbers[0]; // Retrieves the first element (10) int thirdElement = numbers[2]; // Retrieves the third element (30) numbers[1] = 25; // Modifies the second element to 25 Array Length The length of an array in Java, which is the number of elements it can hold can be obtained using the length property: Java int arrayLength = numbers.length; // Returns 5 for the 'numbers' array The length property is a final variable defined in the array object itself and it cannot be changed after the array is created. Iterating Through Arrays Arrays can be traversed using loops such as for or foreach to access and manipulate each element sequentially: Java int[] numbers = {10, 20, 30, 40, 50}; for (int i = 0; i < numbers.length; i++) { System.out.println("Element at index " + i + ": " + numbers[i]); } Alternatively, Java provides an enhanced for-each loop also known as the enhanced for loop to iterate through elements of an array without explicitly using an index: Java for (int number : numbers) { System.out.println(number); } Multidimensional Arrays Java supports multidimensional arrays which are arrays of arrays. we can declare and initialize them as follows: Java dataType[][] arrayName = new dataType[rows][columns]; For example, to create a 2D integer array matrix with 3 rows and 3 columns: Java int[][] matrix = new int[3][3]; Accessing elements in a 2D array requires specifying both row and column indices: Java int element = matrix[1][2]; // Retrieves element at row 1, column 2 Arrays Class Methods The Arrays class in Java provides utility methods for working with arrays such as sorting, searching and comparing arrays: Java import java.util.Arrays; int[] numbers = {5, 3, 8, 2, 7}; Arrays.sort(numbers); // Sorts the 'numbers' array in ascending order int index = Arrays.binarySearch(numbers, 8); // Searches for '8' in the sorted array Other useful methods include copyOf(), fill() and equals(). Common Operations on Arrays Sorting: Arrays can be sorted using Arrays.sort().Searching: Use Arrays.binarySearch() to search for an element in a sorted array.Copying: Arrays can be copied using Arrays.copyOf() or System.arraycopy().Filling: Arrays can be filled with a specific value using Arrays.fill(). Applications of Arrays Arrays are used extensively in various applications, such as: Storing and manipulating collections of data in algorithms and applications.Implementing data structures like lists, queues, and matrices.Handling input/output operations in Java programs.Passing arrays as parameters to methods for processing and manipulation. Conclusion Arrays are fundamental data structures in Java that provide efficient storage and access mechanisms for homogeneous collections of data. They play a crucial role in Java programming offering versatility and performance in managing and manipulating data elements. FAQs 1. What is an array in Java? An array in Java is a fixed-size collection of elements of the same type stored sequentially in memory.2. How do you declare an array in Java? You declare an array in Java by specifying the type of elements followed by square brackets [] and the array name, like int[] numbers;.3. Can arrays in Java store elements of different data types? No, arrays in Java can only store elements of the same data type. Once declared the data type of an array is fixed.4. What is the difference between length and length() in arrays? length is a final variable in arrays that denotes the number of elements it can hold. length() is a method used with the strings and other objects to get the number of characters or elements.5. How do you initialize an array in Java? You can initialize an array in Java using the new keyword followed by the array type and size like int[] numbers = new int[5];.6. What are multidimensional arrays in Java? Multidimensional arrays in Java are arrays of arrays. They allow you to store data in multiple dimensions such as rows and columns in a matrix.7. How can you iterate through an array in Java? we can iterate through an array in Java using a for loop or an enhanced for-each loop to access each element sequentially.8. Can you resize an array in Java once it's created? No, once an array is created with a specific size its size cannot be changed. we would need to create a new array with the desired size and copy elements if resizing is needed.9. What are the common operations you can perform on arrays in Java? Common operations include sorting arrays (Arrays.sort()) searching for elements (Arrays.binarySearch()), copying arrays (System.arraycopy()) and filling arrays (Arrays.fill()).10. What are the applications of arrays in Java? Arrays are used for implementing data structures like lists and queues storing data in algorithms handling input/output operations and passing data to methods efficiently.

Language models become much more useful when they can answer questions about information they were never trained on, including your internal documentation, product manuals, policies, and other proprietary data. Prompting alone cannot solve this, because the model simply does not have access to that knowledge. Retrieval-Augmented Generation, or RAG, is the most common way to bridge that gap. Spring AI comes with solid support for building RAG systems. It has been almost three years since Spring AI showed up, and in that time it has grown from an experimental member of the Spring portfolio into a mature layer over chat models, embedding models, vector stores, and the plumbing that sits between them, which happen to be exactly the pieces a RAG system needs. In this article, we build a small but complete RAG service with Spring AI 2.0. The application reads a set of documents into a PostgreSQL vector store, retrieves the fragments that are relevant to a user question, and lets Anthropic's Claude put together the answer based on those fragments. Everything runs from a standard Spring Boot project, and every step can be reproduced on macOS, Windows, or Linux. The full project is available on GitHub. If you just want to see the finished result, or you would rather skip the step-by-step build below, you can clone the repository and run it as it is. Everyone else can follow along and generate this project from scratch. The prompts themselves are kept deliberately simple. You can tune retrieval and prompts forever; here we care about the architecture and how the pieces fit together in Spring. Approach RAG is not really a single feature. It is more of a small pipeline, and the code below makes a lot more sense once its parts have names. Embedding: a vector of numbers that captures the meaning of a piece of text. Texts that mean similar things end up with vectors that are close to each other.Embedding model: the model that computes these embeddings. It is a different model from the chat model, and it has a different job.Vector store: a database that keeps text fragments together with their embeddings and can answer the question, "which stored fragments are closest in meaning to this query?"Chunking: documents are too large to embed and retrieve as a whole, so we split them into smaller fragments (chunks) before storing them.Similarity search: we embed the user question and fetch the top-k closest chunks from the store.Augmentation: we append the retrieved chunks to the user question before sending it to the chat model, so the model answers from the context we provided instead of from its training data. One thing here is worth calling out, because it shapes the whole setup of the project: the LLM model used in chat and the embedding model are two separate choices. As of today, Anthropic offers LLM models but no embedding API, so a Claude-based RAG system always has to pair Claude with an embedding model from somewhere else. Rather than bringing in a second cloud provider and a second API key, this project computes embeddings locally (inside the JVM), using Spring AI's ONNX transformers module and the well-known all-MiniLM-L6-v2 sentence transformer. It is free and fast enough for this, and it keeps everything on one API key. In our scenario, the service is an internal assistant for a fictional company called Nimbusfield Systems, and it answers employee questions based on the company handbook. The company and the handbook are fictional on purpose. Claude cannot possibly know about it, which makes it easy to verify that the answers really come from our documents and not from the model's own memory. We build this in three steps: Expose a /ask endpoint backed by Claude, with no retrieval, and show that the model cannot answer handbook questions.Ingest the handbook into PGvector at application startup: read, chunk, embed, and store.Attach Spring AI's QuestionAnswerAdvisor to the same ChatClient and ask again. Prerequisites Java 21Maven 3.9.x (the Maven wrapper included in generated projects works too)Spring Boot 4.0.xSpring AI 2.0.0Docker Desktop (macOS/Windows) or Docker Engine (Linux), used only to run PostgreSQL. A project skeleton can be generated at start.spring.io by selecting Web, Anthropic Claude, PGvector Vector Store, and Docker Compose Support. The remaining Spring AI modules are added manually below. The Claude API Key Sign in (or sign up) at the Anthropic Console, open Settings, then API Keys, and create a new key. New accounts may need a small prepaid credit before the API accepts requests, but the runs in this article cost only a few cents. The key is shown only once, so store it right away as an environment variable. If you would rather not spend anything at all, you can still follow along and read through the steps without running the calls yourself. macOS/Linux: export ANTHROPIC_API_KEY=sk-ant-... Windows (PowerShell, persists across sessions after reopening the terminal): setx ANTHROPIC_API_KEY "sk-ant-..." Solution Dependencies With the Spring AI BOM in place, there is no need to repeat versions on the individual artifacts. Initializr expresses the BOM's own version as a property rather than a hardcoded literal, so there is a single place to bump it later: XML <properties> <java.version>21</java.version> <spring-ai.version>2.0.0</spring-ai.version> </properties> <dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-bom</artifactId> <version>${spring-ai.version}</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> A common source of confusion is that start.spring.io has no dependency literally named "Spring AI." Each provider- or store-specific starter (Anthropic Claude, PGvector Vector Database, and so on) is itself a Spring AI module, and picking one transitively pulls in the framework's core classes. (like ChatClient, VectorStore, etc.) Selecting any one of them is also what makes Initializr add the spring-ai-bom as shown above to the generated pom.xml for you. The BOM itself is never a separate item you tick on the Initializr dependency screen. The application needs six Spring AI modules on top of the web starter, each one with a single responsibility. XML <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webmvc</artifactId> </dependency> <!-- Chat model: Anthropic Claude --> <dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-starter-model-anthropic</artifactId> </dependency> <!-- Embedding model: local ONNX sentence transformer --> <dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-starter-model-transformers</artifactId> </dependency> <!-- Vector store: PostgreSQL + pgvector --> <dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-starter-vector-store-pgvector</artifactId> </dependency> <!-- RAG advisor --> <dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-vector-store-advisor</artifactId> </dependency> <!-- Document reading (PDF, Word, Markdown, HTML, and more) --> <dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-tika-document-reader</artifactId> </dependency> <!-- Starts the database container on application startup --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-docker-compose</artifactId> <scope>runtime</scope> <optional>true</optional> </dependency> <!-- Docker Compose service connections for Spring AI vector stores --> <dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-spring-boot-docker-compose</artifactId> <scope>runtime</scope> <optional>true</optional> </dependency> </dependencies> Two models are referenced from the code here. One is the chat model, Claude, which is served from the Anthropic API. The other is the embedding model, which runs locally, right inside the application. We will look at that local embedding model in the next section. The Embedding Model By default, the transformers starter fetches tokenizer.json and model.onnx from Spring AI's own GitHub repository the first time the application starts and then caches them locally. In practice, this default setup is a bit fragile. raw.githubusercontent.com may rate-limit unauthenticated requests, and model.onnx (which is roughly 90 MB) is stored via Git LFS, whose bandwidth quota can run out independently of the ordinary rate limit. When that happens, the endpoint serves the small LFS pointer stub instead of the binary, with a normal-looking HTTP 200, and the failure only shows up later as a cryptic ONNX Runtime protobuf-parsing error rather than a clear download error. The fix is to bundle both files with the application instead of fetching them at startup. So we download them once: Shell mkdir -p src/main/resources/onnx/all-MiniLM-L6-v2 curl -fL -o src/main/resources/onnx/all-MiniLM-L6-v2/tokenizer.json \ https://raw.githubusercontent.com/spring-projects/spring-ai/main/models/spring-ai-transformers/src/main/resources/onnx/all-MiniLM-L6-v2/tokenizer.json curl -fL --http1.1 -o src/main/resources/onnx/all-MiniLM-L6-v2/model.onnx \ https://media.githubusercontent.com/media/spring-projects/spring-ai/main/models/spring-ai-transformers/src/main/resources/onnx/all-MiniLM-L6-v2/model.onnx Then we point the embedding model at these local files in our application.properties, overriding the GitHub-backed defaults: Properties files spring.ai.embedding.transformer.onnx.model-uri=classpath:/onnx/all-MiniLM-L6-v2/model.onnx spring.ai.embedding.transformer.tokenizer.uri=classpath:/onnx/all-MiniLM-L6-v2/tokenizer.json With these two properties set, the application never touches the network for the embedding model, neither on the first run nor on any run after it. The Database The pgvector team publishes a PostgreSQL image with the extension already installed. A compose.yaml in the project root is all we need: YAML services: pgvector: image: "pgvector/pgvector:pg17" environment: - "POSTGRES_DB=nimbusfield" - "POSTGRES_USER=nimbusfield" - "POSTGRES_PASSWORD=nimbusfield" labels: - "org.springframework.boot.service-connection=postgres" ports: - "5432" The labels entry is important. Spring Boot's Docker Compose support auto-detects connection details by matching the image name against a list of well-known images. Plain Postgres is on that list, but pgvector is not, since it is a third-party image. The label tells Spring Boot to treat this container as if it were the official Postgres image, and that is what actually makes the automatic connection wiring work. If we omit it, the container still starts, but Spring Boot never creates a ConnectionDetails bean for it, so the run fails with a connection error rather than falling back gracefully. Because spring-boot-docker-compose is on the classpath, running the application starts the container automatically and injects the connection details. This works the same way on macOS and Windows, as long as Docker Desktop is running. Anyone who prefers to manage the container manually can run the same image with docker run -p 5432:5432 .. and set the datasource properties explicitly. Configuration The complete application.properties, now including the embedding model overrides shown earlier: Properties files spring.ai.anthropic.api-key=${ANTHROPIC_API_KEY} spring.ai.anthropic.chat.model=claude-sonnet-5 spring.ai.anthropic.chat.max-tokens=1024 spring.ai.embedding.transformer.onnx.model-uri=classpath:/onnx/all-MiniLM-L6-v2/model.onnx spring.ai.embedding.transformer.tokenizer.uri=classpath:/onnx/all-MiniLM-L6-v2/tokenizer.json spring.ai.vectorstore.pgvector.initialize-schema=true spring.ai.vectorstore.pgvector.dimensions=384 spring.ai.vectorstore.pgvector.index-type=HNSW spring.ai.vectorstore.pgvector.distance-type=COSINE_DISTANCE logging.level.org.springframework.ai.chat.client.advisor=DEBUG Four details matter here. First, max-tokens is mandatory for the Anthropic API, which caps every response explicitly. Spring AI provides a default, but it is better stated than left implied. Second, the two spring.ai.embedding.transformer.* properties point the embedding model at the local files we bundled in the previous section, instead of Spring AI's own GitHub-backed defaults. See "The Embedding Model" above for why this matters. Third, initialize-schema=true enables the automatic creation of the vector-store table and the required extensions. (Since Spring AI 1.0, this no longer happens silently by default.) Fourth, dimensions=384 must match the embedding model. all-MiniLM-L6-v2 produces 384-dimensional vectors. If the embedding model changes later, the table has to be recreated, because the column type is vector(384). The Documents Two short Markdown files under src/main/resources/docs play the role of the company handbook. remote-work-policy.md Markdown # Nimbusfield Systems Remote Work Policy Employees may work remotely up to three days per week. Remote days must be registered in the portal by Thursday of the preceding week. Working from abroad is permitted for a maximum of 30 calendar days per year and requires prior approval from both the line manager and the People team. travel-expenses.md: Markdown # Nimbusfield Systems Travel and Expenses The daily meal allowance for business trips is 65 EUR in Europe and 80 USD elsewhere. Taxi rides are reimbursed only between airports, hotels, and client sites. Flights longer than six hours may be booked in premium economy. All expense reports are due within 15 working days after the trip via the portal. Thanks to the Tika reader used below, dropping PDFs or Word documents into the same folder works without any code changes. Step 1: Chat Without Retrieval We start with a service that wraps a ChatClient, built once from the auto-configured builder: Java @Service public class AssistantService { private final ChatClient chatClient; public AssistantService(ChatClient.Builder builder) { this.chatClient = builder .defaultSystem(""" You are the internal assistant of Nimbusfield Systems. Answer employee questions precisely and briefly. If you do not know the answer, say so. """) .build(); } public String ask(String question) { return chatClient.prompt() .user(question) .call() .content(); } } And a controller associated with it: Java @RestController public class AssistantController { private final AssistantService assistantService; public AssistantController(AssistantService assistantService) { this.assistantService = assistantService; } @GetMapping("/ask") public ResponseEntity<String> ask(@RequestParam("question") String question) { return ResponseEntity.ok(assistantService.ask(question)); } } Start the application (./mvnw spring-boot:run on macOS/Linux, mvnw.cmd spring-boot:run on Windows) and ask it a handbook question: http://localhost:8080/ask?question=What is the daily meal allowance for business trips in Europe? The response, as we might expect, is: I don't have that information in my available knowledge base. Nimbusfield Systems' specific travel and expense policy—including per diem rates for European business trips—isn't something I can confirm accurately. To get the correct figure, please check: The company's Travel & Expense Policy document (likely on the intranet/HR portal)Your Finance or HR department directlyYour manager, if travel budgets are pre-approved per trip Would you like help with anything else I can assist with more reliably? This gives us a baseline. The model behaves correctly given what it knows, which is nothing at all about this company. Step 2: The Ingestion Pipeline Ingestion follows Spring AI's extract, transform, load structure: a DocumentReader extracts the text, a TextSplitter chunks it, and the VectorStore embeds and stores the chunks. The embedding call happens implicitly inside vectorStore.add() call. The auto-configured TransformersEmbeddingModel is wired into the PgVectorStore and each chunk is embedded into the table. Java @Component public class HandbookIngestion implements ApplicationRunner { private static final Logger log = LoggerFactory.getLogger(HandbookIngestion.class); private final VectorStore vectorStore; private final JdbcTemplate jdbcTemplate; private final Resource[] handbook; public HandbookIngestion(VectorStore vectorStore, JdbcTemplate jdbcTemplate, @Value("classpath:docs/*.md") Resource[] handbook) { this.vectorStore = vectorStore; this.jdbcTemplate = jdbcTemplate; this.handbook = handbook; } @Override public void run(ApplicationArguments args) { Integer count = jdbcTemplate.queryForObject( "select count(*) from vector_store", Integer.class); if (count != null && count > 0) { log.info("Vector store already contains {} chunks, skipping ingestion", count); return; } TokenTextSplitter splitter = TokenTextSplitter.builder() .withChunkSize(300) .build(); for (Resource resource : handbook) { List<Document> documents = new TikaDocumentReader(resource).get(); documents.forEach(doc -> doc.getMetadata().put("source", resource.getFilename())); List<Document> chunks = splitter.apply(documents); vectorStore.add(chunks); log.info("Ingested {} chunks from {}", chunks.size(), resource.getFilename()); } } } The count check makes ingestion idempotent, so restarting the application does not duplicate every chunk. And the source metadata attached to each chunk enables filtered searches later, for instance restricting retrieval to a single document. That same idempotency check has a practical downside worth pointing out. Once the vector store has data, restarting the application will not pick up edits to the handbook files, since the count check short-circuits before the splitter ever runs. To force a clean re-ingestion, for instance after changing a handbook document, tear down the container together with its data volume, not just the container: docker compose down -v The chunk size of 300 tokens is generous for documents this small. The splitter's default of 800 is aimed at larger, real-world content. Chunking is the least exciting and yet the most important knob in a RAG system: chunks that are too large dilute the similarity signals, while chunks that are too small lose their context. It is worth experimenting here: try a few different chunk sizes and see how the system behaves. Just remember to run docker compose down -v between runs, so the vector store is rebuilt from scratch each time. Step 3: Attaching the Retrieval Advisor Now we come back to the plain AssistantService from Step 1 and upgrade it, rather than writing something new. The ChatClient wiring we built earlier stays and what changes is what gets attached to it. Spring AI models the cross-cutting concerns around a chat call as "advisors", which are conceptually close to interceptors. The QuestionAnswerAdvisor embeds the incoming user question, runs a similarity search against the vector store, and appends the retrieved chunks to the prompt before it reaches Claude. Enabling RAG is therefore a change to how the ChatClient is constructed, not to how the request is handled: Java public AssistantService(ChatClient.Builder builder, VectorStore vectorStore) { this.chatClient = builder .defaultSystem(""" You are the internal assistant of Nimbusfield Systems. Answer employee questions precisely and briefly. If you do not know the answer, say so. """) .defaultAdvisors( QuestionAnswerAdvisor.builder(vectorStore) .searchRequest(SearchRequest.builder() .topK(4) .similarityThreshold(0.5) .build()) .build(), new SimpleLoggerAdvisor()) .build(); } topK(4) retrieves at most four chunks per question, and similarityThreshold(0.5) discards weak matches, so an entirely unrelated question augments the prompt with nothing rather than with noise. The SimpleLoggerAdvisor, combined with the DEBUG logging property we set earlier, prints the fully augmented prompt. This is the single most useful debugging tool while tuning retrieval, because it shows exactly what Claude was given. We restart and repeat the same request: http://localhost:8080/ask?question=What is the daily meal allowance for business trips in Europe? The daily meal allowance for business trips in Europe is 65 EUR. Same model, same question, and this time a precise answer grounded in the retrieved handbook chunk instead of a generic deflection. The debug log confirms what is going on behind the scenes: the user question arrives at Claude wrapped in a prompt that contains the retrieved handbook fragments as context. Going Further The default behavior of QuestionAnswerAdvisor is usable, but there are two refinements worth implementing if you want to take this pattern further. The first one concerns grounding. Even with retrieved context, the model may fall back on its general knowledge when the context does not actually contain the answer. The advisor accepts a custom PromptTemplate that controls how the question and the context are merged, and this is the place to enforce stricter behavior. The template must contain the query and question_answer_context placeholders: Java PromptTemplate strictTemplate = PromptTemplate.builder() .template(""" {query} Answer strictly based on the context below. If the context does not contain the answer, reply exactly: "This is not covered by the handbook." --------------------- {question_answer_context} --------------------- """) .build(); QuestionAnswerAdvisor advisor = QuestionAnswerAdvisor.builder(vectorStore) .promptTemplate(strictTemplate) .build(); Asking about, say, the parental leave policy (which is absent from our two files) now produces the fixed refusal instead of an invention. If people are going to rely on it, you want this on. The second refinement could be structured output, and it composes cleanly with retrieval. Declaring a record and calling .entity() instead of .content() gives back a typed object, with Spring AI instructing the model to respond in the matching JSON schema: Java public record HandbookAnswer(String answer, String sourceHint, boolean coveredByHandbook) { } public HandbookAnswer askStructured(String question) { return chatClient.prompt() .user(question) .call() .entity(HandbookAnswer.class); } A last note on the embedding choice. A local MiniLM model is not the strongest embedding model available, and for a large multilingual corpus a hosted embedding API or a bigger ONNX model would retrieve better. This choice is easy to reverse: EmbeddingModel is an interface, swapping the implementation is a matter of a dependency and a property, and the only hard constraint is the one mentioned earlier: the vector dimensions in PGvector have to match whatever the embedding model produces. Conclusion In this article, we built the RAG flow step by step. We started with a plain chat endpoint that could not answer anything about the Nimbusfield handbook, because Claude had never seen it. We then ingested that handbook into PGvector, embedding each chunk locally, and attached Spring AI's QuestionAnswerAdvisor to the same client. That single change was enough to turn a generic model into a service that answers from your own documents. After that, we talked about how we can tighten the grounding, so the model says it does not know when the context has no answer, and pulled the response straight into a typed Java record. If you want to take it further, clone the project, point it at your own documents, apply further the techniques we discussed in the Going Further section, play with different chunk sizes, retrieval settings, and prompts to see how the answers change. The Spring AI documentation goes deeper into advisors, vector stores, and retrieval configuration. The complete, runnable project is available on GitHub.

By Murat Balkan CORE

My first attempt to deploy a Spring Boot microservice on AWS Fargate didn’t fail loudly. It failed quietly — in a loop. ECS kept launching tasks, the Application Load Balancer kept marking them unhealthy, and the service never stabilized. The logs looked fine, the container looked fine, but the ALB replaced every task within seconds. The root cause was painfully simple: Spring Boot needed 45 seconds to start, and my ALB health‑check timeout was 5 seconds. The tasks never had a chance. That night changed how I build and deploy microservices. It forced me to rethink startup behavior, JVM sizing, networking, task definitions, and the entire CI/CD pipeline. This article is the guide I wish I had before that incident — a practitioner’s walkthrough of deploying a production‑ready Spring Boot service on AWS Fargate, with real artifacts and the details that matter when things go wrong. The Architecture That Finally Worked Once the health‑check issue was fixed, the architecture settled into a predictable, cloud‑native flow: Developers push code to GitHubGitHub Actions builds the JARDocker image is built and pushed to Amazon ECRECS service runs AWS Fargate tasksTraffic enters through an Application Load BalancerTasks run in private subnetsConfiguration comes from Parameter Store and Secrets ManagerLogs and metrics flow to CloudWatch It’s the standard modern microservice pipeline — but the difference between “standard” and “production‑ready” is in the details. The Spring Boot Service The microservice itself was simple — a REST API with a few endpoints. The real complexity wasn’t the controller logic; it was everything around it: startup time, health checks, configuration management, and container behavior under load. A Dockerfile Built for Production My first Dockerfile looked like the one many tutorials start with: a single‑stage build running as root with no JVM tuning. It worked locally but failed under real load. Fargate tasks with default JVM heap sizing inside a 2GB container are a classic OOM story. Here’s the hardened version that finally stabilized deployments: Dockerfile FROM eclipse-temurin:21-jre # Create non-root user RUN useradd -u 1001 springuser WORKDIR /app # Layer extraction for faster builds COPY target/*.jar app.jar # JVM tuning for Fargate ENV JAVA_OPTS="\ -XX:MaxRAMPercentage=75 \ -XX:+UseContainerSupport \ -XX:+ExitOnOutOfMemoryError \ " USER springuser ENTRYPOINT ["sh", "-c", "java $JAVA_OPTS -jar app.jar"] This eliminated the OOMKilled events I saw on 2GB tasks and made startup time predictable. Pushing to Amazon ECR With Real Commands The first time I wrote down my ECR commands, they were placeholders. In production, they need to be exact: C aws ecr get-login-password --region us-east-1 \ | docker login --username AWS --password-stdin <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com docker build -t employee-service:1.0.3 . docker tag employee-service:1.0.3 \ <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/employee-service:1.0.3 docker push \ <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/employee-service:1.0.3 Immutable semantic version tags make rollbacks predictable and prevent “latest‑tag roulette.” The ECS Task Definition That Actually Runs in Production A real Fargate deployment lives or dies by its task definition. Here’s the JSON I use today — including secrets pulled from Parameter Store and Secrets Manager: JSON { "family": "employee-service", "networkMode": "awsvpc", "requiresCompatibilities": ["FARGATE"], "cpu": "512", "memory": "1024", "executionRoleArn": "arn:aws:iam::<ACCOUNT_ID>:role/ecsTaskExecutionRole", "taskRoleArn": "arn:aws:iam::<ACCOUNT_ID>:role/employeeServiceRole", "containerDefinitions": [ { "name": "employee-service", "image": "<ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/employee-service:1.0.3", "portMappings": [ { "containerPort": 8080, "protocol": "tcp" } ], "secrets": [ { "name": "DB_PASSWORD", "valueFrom": "arn:aws:ssm:us-east-1:<ACCOUNT_ID>:parameter/db/password" }, { "name": "API_KEY", "valueFrom": "arn:aws:secretsmanager:us-east-1:<ACCOUNT_ID>:secret:thirdparty/api" } ], "logConfiguration": { "logDriver": "awslogs", "options": { "awslogs-group": "/ecs/employee-service", "awslogs-region": "us-east-1", "awslogs-stream-prefix": "ecs" } } } ] } The ALB Health Check That Stopped the Outage My outage happened because the ALB was impatient. Here’s the configuration that finally stabilized deployments: settingvalue Path /actuator/health Interval 20 seconds Timeout 10 seconds Healthy threshold 3 Unhealthy threshold 3 Spring Boot startup time + ALB patience = stable deployments. Why Fargate Tasks Belong in Private Subnets Early on, I deployed tasks in public subnets because it felt simpler. It wasn’t. Public IPs meant the containers were directly reachable from the internet — port scans, bot traffic, and noisy logs. Moving tasks to private subnets solved several problems at once: Reduced Attack Surface No public IPs. No direct inbound traffic. Only the ALB can reach the tasks. A Single Secure Entry Point The ALB handles TLS termination, redirects HTTP→HTTPS, performs health checks, and integrates with WAF. Clients never bypass it. Cleaner Security Groups ALB SG: inbound 443 from the internetTask SG: inbound only from ALB SG Nothing else touches the containers. Compliance Alignment PCI, SOC 2, HIPAA — all prefer minimizing public exposure. Controlled Outbound Access Tasks use a NAT Gateway for outbound calls (updates, third‑party APIs) without exposing themselves. Better Scalability ALB target groups automatically track tasks across AZs as ECS scales. The architecture becomes simple and predictable: Internet → ALB (public subnets) → Fargate tasks (private subnets) It’s quieter, safer, and easier to operate. The GitHub Actions Workflow That Deploys Automatically Here’s the pipeline that builds, tests, pushes, and deploys the service: YAML name: Deploy to Fargate on: push: branches: ["main"] jobs: build-deploy: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Set up JDK uses: actions/setup-java@v4 with: java-version: "21" - name: Build JAR run: mvn -B clean package - name: Login to ECR uses: aws-actions/amazon-ecr-login@v2 - name: Build and Push Image run: | docker build -t employee-service:1.0.3 . docker tag employee-service:1.0.3 ${{ env.ECR_REGISTRY }/employee-service:1.0.3 docker push ${{ env.ECR_REGISTRY }/employee-service:1.0.3 - name: Deploy ECS Service uses: aws-actions/amazon-ecs-deploy-task-definition@v2 with: task-definition: ecs-task.json service: employee-service cluster: prod-cluster Auto Scaling With Real Target Tracking JSON Target tracking is the simplest and most reliable scaling strategy for Fargate: JSON { "TargetValue": 50.0, "PredefinedMetricSpecification": { "PredefinedMetricType": "ECSServiceAverageCPUUtilization" }, "ScaleOutCooldown": 30, "ScaleInCooldown": 60 } I use 50% as the target because it balances cost and responsiveness. What I Learned Every failure taught me something: ALB timeouts taught me to respect startup timeOOMKilled tasks taught me to tune the JVMPublic subnets taught me to isolate workloadsManual deployments taught me to automate everything AWS Fargate really does deliver on its promise — no servers to manage, automatic scaling, and clean integration with ECS — but only after you learn the hard parts. If you’re deploying Spring Boot on Fargate, I hope you learn those lessons from this article instead of from your own outage.

Cross-border e-commerce sellers often spend hours comparing the same products across different Amazon marketplaces. Prices, reviews, and seller signals vary by country, but the process is still largely manual. I wanted to see how far I could automate it with a small AI agent built using Codex, SerpApi, and Lark. Suppose I ask in Codex: Compare Grogu products in the US and Japan The workflow is straightforward: Search Amazon marketplacesFetch product detailsSummarize seller signalsSend a report to Lark I kept the stack intentionally simple: Python 3.12uvPydanticSerpApi Python SDKOpenAI structured outputsLark custom bot webhookCodex for both development and as the conversational interface Running the Agent From Codex Once everything was wired together, I could simply ask Codex: Compare Grogu products in the US and Japan Codex understood the request, triggered the CLI workflow, collected marketplace data via SerpApi, and delivered a structured report to Lark. Here’s the entire flow in action: The Workflow The overall flow looks like this: Starting with a natural-language request in Codex, the agent first validates that the question is related to cross-border Amazon research. OpenAI then translates the request into a structured command. SerpApi handles both product discovery and detailed product retrieval, while OpenAI extracts seller-focused insights from the collected data. Finally, the results are packaged into a Lark card and delivered via a Custom Bot webhook. I intentionally did not start with FastAPI, background jobs, or a database. For the MVP, the important question was simpler: Can I go from a natural-language product question to a useful cross-border product card? Translating Natural Language into Commands The natural-language entry point uses OpenAI to translate user requests into structured commands. For example: Compare Grogu products in the US and Japan becomes: YAML query="Grogu" marketplaces=["us","jp"] output_mode="send_lark" Using a strict schema keeps the pipeline predictable and much easier to debug. Instead of letting the model orchestrate everything, I only ask it to generate structured commands. Searching Amazon With SerpApi This project relies on two SerpApi endpoints. The Amazon Search API is used to discover candidate products, while the Amazon Product API enriches them with much richer details. Search results provide ranking context and thumbnails, while product pages contain detailed information such as images, availability, and descriptions. Combining both produced much better product cards than using either endpoint alone. Modeling Product Data Amazon pages are messy. Some products have ratings but no availability. Some have images but no variants. Some fields simply don't exist. Missing data is normal, not an exception. My first instinct was to say, "Why not model most fields as optional?" That is true, but it is not the whole solution.The real issue was that SerpApi returns useful product data from several different places. Some fields are flat. Some fields are nested. Some fields have different names depending on whether they came from Amazon Search API or Amazon Product API. Some products should not be shown at all if they are missing the signals a seller actually needs. To make the data usable, I: Normalized both endpoints into a common Product model;Filtered out products with weak seller signals;Merged Product API details back into search results;Treated missing fields as expected rather than failures. The Product model still uses optional fields, because missing data is normal: But optional fields alone were not enough. I also filtered search results before choosing products for detail lookup: This was important for the Lark card. A cross-border seller does not want a table full of Rating: N/A and Reviews: N/A. Those rows make the card noisy and less actionable. The next issue was nested and inconsistent JSON. For price, SerpApi may return a string, a number-like value, or a nested object: For availability, the Product API does not always use one stable field. I had to check availability, stock, and sometimes delivery: Keeping the Agent Narrow The most interesting part of the project isn't Grogu. It's scope. The entry point only supports cross-border Amazon product research. If somebody asks: What is today's weather? the app simply rejects the request. Requests outside the project scope are rejected locally before calling OpenAI. For supported requests, OpenAI returns a strict command object: This keeps the agent predictable. It translates requests into commands instead of improvising actions. I deliberately avoided turning this into a general chatbot. The agent only knows one workflow: cross-border Amazon product research. That narrow scope makes the behavior easier to explain, test, and trust. It also keeps OpenAI API responsible for translation rather than improvisation. Generating Seller-Friendly Insights and Delivering Them to Lark Once product data has been collected, OpenAI generates seller-focused insights. I intentionally constrain the model to use only information returned by the APIs, avoiding hallucinated prices, ratings, or availability. Instead of producing generic summaries, the analysis focuses on demand signals, social proof, pricing, and obvious risks — information that is much more useful for sellers. The final result is delivered through a Lark Custom Bot webhook. Error Handling External APIs fail. That's normal. I treated each layer independently. If OpenAI fails, no command is generated. If SerpApi fails, the analysis stops with a clear message. If Lark delivery fails, the report can still be viewed locally. This separation keeps failures localized and prevents one component from bringing down the entire workflow. Conclusion The most interesting lesson from this project wasn’t the Grogu theme or the Lark card. It was learning where the boundaries should be. The agent works because it stays narrow. That narrowness makes the system easier to understand, debug, and trust. And if you’re building tools for cross-border e-commerce, SerpApi’s Amazon API provides a surprisingly rich source of product data. They made this entire workflow possible. If you’re working on product research, seller analytics, or marketplace intelligence, I’d recommend giving them a try. Check out the full SerpAPI article collection here.

Most enterprise data problems are not caused by machine learning models or dashboard tools. They usually start much earlier in the pipeline. A reporting table misses records after a schema change. A nightly ingestion job finishes successfully but loads duplicate transactions. A downstream dashboard suddenly shows a 30% increase in revenue because one transformation joined datasets incorrectly. These issues are common in large-scale analytics environments where pipelines evolve faster than governance processes. PySpark is often adopted because it can process large distributed workloads efficiently, but scalability alone does not guarantee reliability. In practice, many pipelines become difficult to debug, validate, and maintain as the amount of data and its transformation complexity increase. This article focuses on practical techniques for building reliable analytics pipelines with PySpark, which include validation strategies, transformation design, partition management, schema handling, and operational monitoring. Reliability Problems in Enterprise Pipelines Pipeline failures rarely happen because Spark cannot process the data. They happen because the surrounding engineering practices are weak. One common issue is silent schema drift. A source system adds a new column or changes a data type, and downstream transformations continue running without immediately failing. The pipeline technically succeeds, but the analytics layer begins producing incorrect results. Another common problem appears during joins. Large transactional datasets often contain duplicate business keys, incomplete reference mappings, or delayed records. A transformation that works correctly during testing may suddenly inflate row counts in production. Operational reliability also becomes harder when transformations are tightly coupled. In many environments, a single failed stage forces the entire workflow to rerun, increasing compute costs and delaying reporting cycles. These problems are usually not visible during initial development. They appear after pipelines begin handling larger workloads, inconsistent source systems, and changing business logic. Structuring Pipelines for Maintainability One mistake teams make early is treating PySpark jobs as large monolithic scripts. That approach works temporarily, but maintenance becomes difficult once pipelines grow beyond a few transformations. Small schema changes become risky because logic is spread across multiple dependent stages. A more maintainable design separates the pipeline into distinct layers: IngestionValidationTransformationEnrichmentAggregationOutput The separation matters because each layer serves a different operational purpose. For example, ingestion should preserve source fidelity as much as possible. Validation layers should isolate problematic records before transformations begin. Aggregation logic should not contain ingestion-specific assumptions. This layered approach makes debugging significantly easier during production incidents. Validation Should Happen Early Many pipeline implementations validate data too late. Teams often begin transformations immediately after loading raw datasets, assuming upstream systems already enforce quality controls. In reality, enterprise data sources frequently contain null business keys, inconsistent timestamps, malformed identifiers, and duplicate records. Validation becomes much more manageable when it happens near ingestion. At minimum, validation checks should include: Null business keysDuplicate identifiersDatatype consistencyTimestamp integrityUnexpected categorical valuesRow count anomalies A practical pattern is separating invalid records into quarantine datasets instead of failing the entire pipeline immediately. This prevents a small number of bad records from interrupting large scheduled workflows while still preserving visibility into data quality issues. Another useful technique is maintaining row-count checkpoints between stages. Unexpected increases or decreases often identify join problems much faster than reviewing transformation logic manually. Managing Expensive Transformations Performance issues in PySpark pipelines usually come from unnecessary shuffling and poor partition strategies rather than raw data volume. Joins are one of the biggest causes of instability in large pipelines. A transformation that performs adequately in development can become extremely expensive once dataset sizes increase. Partitioning strategy matters here. Over-partitioning creates scheduling overhead and small files. Under-partitioning causes skewed workloads where a few executors process most of the data while others remain idle. The challenge is that there is no universally correct partition count. Pipelines behave differently depending on: Dataset cardinalityJoin distributionFile sizesCluster configurationTransformation complexity This is why reliable Spark pipelines require continuous observation and tuning rather than static optimization rules. Caching also needs careful handling. Many teams cache aggressively without monitoring executor memory pressure, eventually degrading overall cluster performance instead of improving it. In practice, caching should only be used for reused intermediate datasets with measurable recomputation costs. Handling Schema Evolution Safely Schema evolution becomes unavoidable in long-running analytics environments. New attributes are introduced. Legacy fields are deprecated. Source applications modify export structures without warning. Pipelines that rely on rigid assumptions eventually fail under these conditions. One practical approach is maintaining explicit schema contracts between ingestion and transformation layers. Instead of relying entirely on inferred schemas, pipelines should validate expected columns and datatypes before downstream processing begins. Backward compatibility also matters. Adding nullable fields is usually manageable. Renaming or changing datatypes is far riskier because downstream dependencies may silently break. This becomes especially important when multiple teams consume the same datasets. Reliable pipelines are not only technically correct; they are predictable for downstream consumers. Observability Is More Important Than Most Teams Realize Many Spark jobs are difficult to troubleshoot because they produce very little operational metadata. A successful pipeline should generate more than output files. Useful operational metrics include: Processed row countsRejected row countsStage execution durationPartition statisticsFreshness indicatorsSchema validation failures Without observability, debugging production incidents becomes reactive and slow. Logging also needs structure. Generic console output is rarely sufficient once pipelines become distributed across multiple workflows and orchestration systems. This is one reason mature analytics environments invest heavily in monitoring frameworks and lineage tracking systems. The goal is not simply running pipelines successfully. The goal is understanding pipeline behavior consistently over time. Reliability Requires Engineering Discipline PySpark provides distributed processing capabilities, but reliable analytics systems depend far more on engineering discipline than framework selection. Many pipeline failures are preventable: Transformations without validationUnmanaged schema changesMissing operational metricsTightly coupled workflowsInconsistent partitioning strategies As analytics environments scale, reliability becomes increasingly important because downstream systems depend on pipeline consistency for reporting, forecasting, machine learning, and operational decision-making. Teams that prioritize reliability early usually spend less time firefighting production issues later. Conclusion Reliable data pipelines are foundational to enterprise analytics, yet reliability is often treated as a secondary concern until failures begin affecting downstream reporting and operational workflows. PySpark provides the scalability needed for modern analytics workloads, but scalable pipelines are not automatically reliable. Reliability comes from disciplined validation practices, careful transformation design, observability, and operational maintainability. As organizations continue expanding their analytics capabilities, pipeline engineering quality increasingly determines whether downstream insights can actually be trusted.