Data Resources

DZone's Featured Data Resources

Microsoft Fabric Throttling: Diagnose Before You Scale

If Your Fabric Capacity Is Throttling, Do Not Start With the SKU A report takes twenty seconds to open. A pipeline that normally finishes before breakfast is still running. Someone sees the capacity above 100% and proposes the obvious fix: move from F64 to F128. That might be the right decision. It should not be the first one. Microsoft Fabric brings data engineering, warehousing, real-time analytics, data science, and Power BI onto shared capacity. That is one of its strengths. It also means one inefficient item can consume compute that another workload needs. A poorly shaped warehouse query, an oversized Spark session, an hourly refresh of data that changes once a day, or a report page with too many expensive visuals can all surface as the same complaint: “Fabric is slow.” The complaint does not tell you the cause. The useful operating model is: Confirm the capacity is throttling. Isolate the operation creating pressure. Optimize that workload. Scale only when the remaining demand is legitimate. This distinction matters because a larger SKU gives inefficient work more room to run. It does not make the work efficient. First, Separate Slow Performance From Throttling Slow performance and capacity throttling can look similar to a user, but they are different problems. A report visual can be slow because its DAX is expensive. A warehouse query can be slow because it scans far more data than it needs. A Spark job can wait because it requested more executors than are available. Network latency, gateway pressure, cold cache, locks, and source-system performance can also add delay. Throttling is more specific. It happens when billable operations consume more Capacity Unit seconds than the capacity can supply, including the future capacity Fabric uses to absorb smoothed usage. When that budget is exhausted, Fabric progressively delays or rejects new work. This leads to the first rule: Do not diagnose throttling from a slow screen or a utilization spike. Confirm it in the Capacity Metrics app. Microsoft explicitly notes that utilization above 100% does not immediately mean throttling. Fabric allows bursting, smooths usage, and provides limited overage protection before user operations are affected. Use the Throttling view and System events on the Compute page to confirm whether the capacity entered interactive delay, interactive rejection, or background rejection. Rejected requests can also surface errors such as CapacityLimitExceeded. If the app shows no throttling during the incident, investigate the item itself. Buying capacity to fix a bad query is an expensive way to avoid reading its execution plan. What Fabric Is Protecting You From Fabric does not force every operation to fit inside the CU budget of the current second. It uses two mechanisms to absorb variability. Bursting lets an operation temporarily use more compute than the provisioned SKU provides so it can finish quickly. Smoothing spreads the CU accounting for that operation across future 30-second timepoints. Interactive operations are typically smoothed over five to 64 minutes, depending on consumption. Background operations are typically smoothed over 24 hours. Smoothing is not a performance optimization. It does not make a query cheaper or slow a job down to fit the capacity. It changes how consumed compute is applied against the capacity limits. When smoothed usage continues to accumulate faster than idle capacity can pay it down, Fabric moves through progressive protection stages: Future capacity consumedProtection stageUser impactUp to 10 minutesOverage protectionNew operations continue without throttlingMore than 10 minutes, up to 60 minutesInteractive delayNew interactive operations receive a 20-second submission delayMore than 60 minutes, up to 24 hoursInteractive rejectionNew interactive operations are rejectedMore than 24 hoursBackground rejectionNew interactive and background operations are rejected Operations already running are allowed to complete. The capacity recovers when unused CUs burn down the accumulated carryforward. That delayed accounting explains a common mystery. The item users see failing at 10:00 AM might not be the item that created the pressure at 10:00 AM. A large background operation completed earlier can still be consuming future capacity through smoothing. Diagnose From the Capacity Down, Not From the complaint up When a capacity incident occurs, use the same investigation path every time. 1. Pin Down the Affected Window Record the time, capacity, workspace, reported symptom, and whether the failure affected interactive work, background work, or both. Do not begin with a full month of data. Start with the narrowest window that contains the incident. The Capacity Metrics app shows the previous 14 days of compute usage and displays hourly usage in the multimetric ribbon. Select the affected day and hour. When filters are applied, the utilization and throttling visuals can show 30-second timepoints rather than only periodic peaks. 2. Confirm the Protection Stage Open the Throttling tab and check which threshold crossed 100%: Interactive delay means the capacity consumed more than 10 minutes of future capacity.Interactive rejection means it consumed more than 60 minutes.Background rejection means it consumed more than 24 hours. Then review Overage (Carryforward). The add, burndown, and cumulative values tell you whether the capacity experienced a short surge that quickly recovered or sustained demand that kept adding debt. Minutes to burndown is especially useful. It estimates how long recovery would take if no new operations entered the capacity. 3. Find the Items That Consumed the CUs In the matrix by item and operation, sort by CU(s) for the selected window. Expand the highest consumers to see the operation type. Duration alone is not enough. A long-running operation can use modest compute, while a short burst can consume many CUs. Add the optional columns that help explain the incident: Rejected countFailed countSuccessful countIn-progress countOverloaded minutesItem sizePerformance delta Hovering over matrix values exposes the operation-level breakdown. Capture the workspace, item, operation, user or service principal, CU seconds, operation count, and time. 4. Classify the Shape of the Problem Most incidents fall into one of four patterns: PatternWhat the evidence looks likeLikely responseOne expensive operationA single run dominates CU secondsOptimize code, query, model, or data layoutExcessive frequencyA modest operation runs far more often than its data changesReduce refresh or execution frequencyHigh concurrencyMany acceptable operations overlapControl concurrency, isolate workloads, or scaleSustained business demandOptimized work remains consistently highScale up or scale out This classification prevents random tuning. It tells you whether to reduce cost per operation, reduce operation count, change where the work runs, or provide more compute. Reducing frequency and rescheduling are not the same optimization. Changing an hourly refresh to a daily refresh removes unnecessary executions and lowers total consumption. Moving the same 24 hourly refreshes to a different part of the day does not make them cheaper. It only changes when they compete with other work. Optimize the Workload That Actually Caused the Pressure Fabric workloads share capacity, but they do not share one tuning playbook. Once you know which operation consumed the CUs, focus on that workload instead of trying to optimize everything. Warehouse and SQL Analytics Endpoint Start with Query Insights to identify completed queries using the most CPU. Microsoft's Query Insights documentation includes this query: SQL SELECT TOP 100 distributed_statement_id, query_hash, allocated_cpu_time_ms, label, command FROM queryinsights.exec_requests_history ORDER BY allocated_cpu_time_ms DESC; If a query stands out, check how much data it scans, whether filters are applied early, and whether joins or statistics need attention. Compare repeated executions, not just the first run, because cold starts can distort the picture. The goal is fewer resources consumed for the same result, not simply a faster first response. Spark, Notebooks, and Lakehouse Processing Use Spark monitoring and the Tune tab to look for heavy shuffles, skewed stages, spills, or executors that sit underused. Filter data early, avoid repeated actions, and review Delta file layout when small files become a recurring problem. Also check session lifetime. Leaving compute running after the work is done is easy to overlook. If bursty Spark workloads regularly compete with reporting, consider whether separately billed on-demand Spark is a better fit. Measure the cost and isolation trade-off before making the switch. Data Factory Pipelines and Dataflow Gen2 A pipeline can orchestrate several expensive operations, so the pipeline itself may not be where the CUs are going. Check the notebook, copy, refresh, or transformation activity behind it. The simplest savings often come from removing unnecessary runs, using incremental loads, and preserving query folding in Dataflow Gen2 where supported. Moving the same work to another hour might reduce contention, but it does not reduce total CU consumption. Removing work you do not need does. Power BI Semantic Models and Reports Separate refresh pressure from interactive report pressure. For refreshes, focus on incremental processing, folding, and schedules that match how often the source changes. For reports, use Performance Analyzer to find expensive visuals and DAX measures, then compare those findings with the semantic model's operations in Capacity Metrics. A page with too many costly visuals can feel like a capacity problem even when the better fix is in the report design. Real-Time Intelligence For KQL databases and Eventhouses, inspect wide queries, repeated aggregations, retention and caching choices, and ingestion patterns. Limit scans to the data actually needed and consider pre-aggregation when queries repeatedly calculate the same result. Keep the freshness requirement in view. Lower CU consumption is not a win if the data now arrives too late to be useful. Real-Time Intelligence also has different throttling behavior, so do not assume every workload uses the same interactive delay stage. Reduce Blast Radius When Optimization Is Not Enough Some workloads are efficient and still should not share a failure boundary. An executive reporting estate, a self-service experimentation workspace, and a large overnight engineering platform can have different service expectations. If one workload is allowed to consume the headroom needed by another, the architecture is coupling them operationally even if their data products are separate. Fabric gives you several options: Scale up when optimized demand legitimately exceeds the current SKU.Scale out by moving workspaces to another capacity when isolation, administration, or service levels differ. This adds capacity cost and operational ownership, so the service-level benefit must justify it.Use on-demand billing for Spark when dynamic Spark demand should not consume the shared capacity.Configure surge protection to limit the background compute that can consume capacity and make interactive throttling less likely.Enable capacity overage billing when continuity is worth the additional cost and the feature fits your operating model. Microsoft documents overage billing at three times the normal capacity rate, so estimate the cost first. Surge protection is a guardrail, not free capacity. When it activates, background jobs can be rejected. Microsoft recommends correctly sizing and isolating critical solutions rather than treating surge protection as a replacement for optimization or capacity design. Prove That the Optimization Worked A workload is not optimized because its code changed or one test completed faster. Compare the same business operation before and after the change. Track: MeasureWhy it mattersCU seconds per operationShows whether the work became cheaperOperation countExposes unnecessary frequency and retriesTotal CU secondsShows the net capacity effectDurationMeasures user or batch completion timeData scannedReveals pruning, cache, and query-shape improvementsRejected and failed countConfirms whether capacity pressure still disrupts workCarryforward and minutes to burndownShows whether the capacity recovers fasterInteractive latencyConfirms that users experience the improvement Run the comparison over a representative period. A change that improves one query but shifts expensive work into refresh, maintenance, or another engine is not an end-to-end optimization. For recurring operations, calculate both cost per run and daily cost: Plain Text Daily CU-seconds = CU-seconds per operation × operations per day Reducing a query from 1,000 to 800 CU seconds saves 20% per run. If it runs ten times more often after the redesign, total consumption still increases substantially. Efficiency and frequency belong in the same calculation. A Practical Throttling Response Worksheet Use this during the next capacity incident: Plain Text Incident window: Capacity and SKU: User-visible symptom: Confirmed throttling stage: Carryforward and minutes to burndown: Top workspace: Top item: Top operation: Interactive or background: CU-seconds: Operation count: Rejected count: Problem shape: [ ] One expensive operation [ ] Excessive frequency [ ] High concurrency [ ] Sustained business demand Optimization applied: Before CU-seconds per operation: After CU-seconds per operation: Before daily CU-seconds: After daily CU-seconds: Problem shifted to refresh, maintenance, or another engine: Decision: [ ] Keep current SKU [ ] Scale up [ ] Isolate on another capacity [ ] Move Spark to on-demand billing [ ] Add a protective capacity control This turns a vague performance complaint into a traceable engineering decision. The Goal Is Not Zero Utilization A well-used Fabric capacity should do work. The goal is not to keep every chart comfortably low or eliminate every temporary excursion above 100%. The goal is to ensure that each operation earns the compute it consumes, critical workloads retain the service level they need, and sustained business demand fits inside a deliberate capacity envelope. When throttling appears, ask these questions in order: Did the capacity actually throttle, or is one item simply slow?Which item and operation consumed the CUs?Is the problem cost per operation, frequency, concurrency, or real growth?Can the workload be made cheaper without reducing the business outcome?If not, should the answer be more capacity or better isolation? That is the difference between reacting to throttling and operating Fabric well. Sources This article is grounded in current Microsoft documentation. Product behavior and preview features can change, so verify the latest guidance before making production capacity decisions. Understand capacity throttling and smoothing (Microsoft Learn)Capacity Metrics app compute page (Microsoft Learn)Evaluate and optimize your Microsoft Fabric capacity (Microsoft Learn)Performance guidelines for Fabric Data Warehouse (Microsoft Learn)Query Insights in Fabric Data Warehouse (Microsoft Learn)On-demand billing for Apache Spark in Microsoft Fabric (Microsoft Learn)Optimization guide for Power BI (Microsoft Learn)Manage surge protection for Fabric capacities (Microsoft Learn)Data clustering in Fabric Data Warehouse (preview) (Microsoft Learn)Partitioning for Delta tables (Microsoft Learn) More

Why Spark Executors Leave Cores Idle on YARN: A Packing Model for Executor Sizing

By Vikrant Bachhav

The Problem A batch Spark job on YARN can show memory nearly full and CPU half idle at the same time, and you pay for both. Memory looks full because YARN reports the memory allocated to containers, not the memory in use. CPU looks idle because the executors that fit on each node don't add up to the node's cores. The cluster is billed for every vCPU and uses a fraction of them. Background: How Spark Sizes Itself on YARN Spark requests one YARN container per executor [3]. The container size is the executor heap (spark.executor.memory) plus off-heap overhead (spark.executor.memoryOverhead), which defaults to 10% of executor memory with a 384 MiB minimum [1]. The heap divides into a small reserved area, a unified region shared by execution and storage (60% of the usable heap by default, set by spark.memory.fraction), and the remainder for user data structures [2]. YARN also rounds each request up to its allocation increment, so real fits can differ slightly from paper math [4]. Two limits decide how many executors land on a node: the vCPUs and the memory that the NodeManager advertises. Whichever runs out first sets the count. Python executors_per_node = min( floor(node_vcores / cores_per_executor), floor(node_memory_mb / ((executor_memory_gb + overhead_gb) * 1024)) ) The Packing Model in Practice Take a worker that exposes 16 vCPUs and 112 GB (114,688 MB) to YARN, running executors with 5 cores each. By cores alone, the node fits three executors. Memory decides whether it actually does. Executor memory + overheadFit by memoryFit by coresExecutors per nodeBusy vCPUsCore utilization64 + 2 GB131531%48 + 2 GB2321062.5%32 + 1 GB3331594%24 + 1 GB4331594% The first two rows are the trap. A large executor looks generous, yet it strands cores that you still pay for. The third row is the sweet spot: the core limit and the memory limit meet. The last row shows that shrinking memory further gains nothing, because the core limit now binds. The overhead setting changes the packing too. With the default 10% factor, a 48 GB executor requests 4.8 GB of overhead. Setting it explicitly to a smaller value can raise the number of executors per node, and setting it too low can get containers killed for exceeding their memory limit. The misleading dashboard follows from the same model. YARN's memory-used metric counts allocated container memory. A node full of reservations looks 100% used even when the executors touch only part of their heap. High memory use plus low CPU use is therefore a signature of oversized executors, not a contradiction. The Approach The fix is to shrink the executor until the core limit binds instead of the memory limit. Keep the cores per executor fixed so that memory is the only variable. Five cores is the common starting point, a heuristic that grew out of HDFS client throughput [7][8]. Leave one or two vCPUs per node free for the NodeManager and the operating system. Then set the shuffle partition count. The default of 200 leaves a large cluster mostly idle during shuffles. A common rule is 2 to 3 times the total core count, checked against a partition size near 128 MB. When adaptive query execution is on, which is the default from Spark 3.2, Spark can coalesce small shuffle partitions at runtime, and the configured number sets the ceiling [6]. Plain Text spark.executor.cores 5 spark.executor.memory 32g spark.executor.memoryOverhead 1g spark.sql.shuffle.partitions 4800 # about 3x total cores (1,600 in this example) spark.dynamicAllocation.enabled true spark.shuffle.service.enabled true spark.speculation true spark.speculation.multiplier 3 spark.speculation.quantile 0.9 spark.executor.extraJavaOptions -XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35 A few of these deserve a comment. Executor autoscaling on YARN is usually paired with the external shuffle service so that shuffle files outlive their executors. Speculation relaunches tasks that run more than three times slower than the median once 90% of a stage's tasks have finished, which trims straggler tails at the price of some duplicate work. The G1 occupancy threshold of 35 starts concurrent marking earlier than the default of 45, trading some CPU for a lower risk of full garbage collections on large heaps. The Failure Mode: Local Disk Fixing CPU utilization can expose a limit you never reached before. More active cores per node means more concurrent shuffle writers on the same local disks. When a disk passes the NodeManager health checker's utilization threshold, which defaults to 90% and applies to both the local and log directories, YARN marks the node unhealthy and stops scheduling on it [4][5]. In a bad case, a large share of the cluster drops out mid-job, and the run has to be killed. The remedies are mechanical. Give workers larger local disks, size shuffle partitions so each one spills less, and watch the unhealthy-node count during any resizing experiment. Run the experiment at the quietest and busiest hours of the day, because the busy hour is where disk pressure shows up. Query-Level Tuning Still Matters Cluster sizing only goes so far. Find data skew first, since it is the most common cause of stuck stages. Prefer broadcast joins when one side is small enough to fit in executor memory, use equi-joins where possible, and always filter on the partition key. Look for joins between two very large datasets that a pre-aggregation could replace. Removing one such join can shrink the cluster a job needs far more than any configuration change. Trade-Offs Smaller executors fit more tasks per node, but each task gets less memory. Skewed tasks are more likely to hit out-of-memory errors, and broadcast variables are copied once per executor. Larger executors share broadcast data across more cores and carry less per-JVM overhead, but they can strand cores and tend to have longer garbage collection pauses. More active cores per node means more concurrent shuffle writes, so size local disks for the peak and not the average. The five-core rule is a heuristic. Test it against your own storage layer, whether that is HDFS or an object store. A Sizing Routine You Can Reuse Read the NodeManager memory and vCore limits from the ResourceManager UI. Fix the cores per executor. For each candidate memory size, compute executors per node with the function below. Pick the largest memory size that still reaches the core limit. Set shuffle partitions to 2 to 3 times the total cores, then check the partition size. Test at a quiet hour and at the peak, and watch NodeManager disk health. Python def executors_per_node(node_vcores, node_mem_mb, cores_per_exec, exec_mem_gb, overhead_gb): by_cpu = node_vcores // cores_per_exec by_mem = node_mem_mb // int((exec_mem_gb + overhead_gb) * 1024) return min(by_cpu, by_mem) for mem, ovh in [(64, 2), (48, 2), (32, 1), (24, 1)]: n = executors_per_node(16, 114688, 5, mem, ovh) print(mem, ovh, n, n * 5) # executors per node, busy vCPUs The Result Applied to a production batch job, this model explained a large gap between the cores the cluster paid for and the cores it used. Resizing the executors moved the busy cores close to the node's usable limit and shortened the run. The same job later needed fewer workers. The practical rule is simple: count the containers that fit on a node before changing any setting. Vikrant Bachhav is a Staff Software Engineer specializing in large-scale data infrastructure, real-time and batch ML feature pipelines, and cloud cost optimization. References [1] Apache Spark, Configuration [2] Apache Spark, Tuning Spark [3] Apache Spark, Running Spark on YARN [4] Apache Hadoop, yarn-default.xml [5] Amazon EMR, Cluster terminates with NO_SLAVE_LEFT and core nodes FAILED_BY_MASTER [6] Databricks, Adaptive Query Execution: Speeding Up Spark SQL at Runtime [7] Distribution of Executors, Cores and Memory for a Spark Application [8] PayPal Tech Blog, Tuning Spark Applications to Efficiently Utilize Dataproc Cluster More

Designing Memory Architecture for Stateful AI Agents: Short-Term Context and Long-Term Persistence

By Kapil Shirish Bidikar

Introduction to Error Prone: How to Catch Bugs at Compile Time

By Hüseyin Akdoğan

CORE

Why Vulnerability Management Fails Without Reliable Asset Data

By Alex Vakulov

CORE

LangChainGo vs Genkit Go: Where Genkit Shines

There are two realistic choices for building AI services in Go today: LangChainGo, the Go port of the Python framework everyone knows, and Genkit Go, Google's open-source Gen AI framework. LangChainGo is still the first result most Go developers find. It has 9,677 stars and the biggest provider catalog in the ecosystem. It is also eight months without a commit, eleven months without a release, and still on 0.1.x after three and a half years. This article compares them the way that actually matters: by building the same production service in both and looking at what each one costs you. Not a feature checklist, a real application with real requirements. If you have already decided and want the migration mechanics, that is the companion article: How to migrate from LangChainGo to Genkit Go. The State of Both Projects LangChainGoGenkit GoLatest releasev0.1.14, October 2025v1.13.1Last commit to mainJanuary 2026activeAge and version3.5 years, still 0.1.xGA since v1.0Open issues242Open pull requests171, oldest from June 2023Stars9,677 LangChainGo is not archived, and "dead" overstates it. But eleven months without a release is long enough for the ecosystem to move underneath a library, and it has. What Is Broken in LangChainGo Today Three concrete things, all of which affect a service in production right now. The quickstart returns a 404. DefaultOptions() hardcodes gemini-2.0-flash, which Google retired. A fresh googleai.New(ctx) fails on the first call. The Google provider runs on an end-of-life SDK. llms/googleai imports github.com/google/generative-ai-go, whose README states support ended on November 30, 2025. Streaming is broken on every current Gemini model. gemini-2.5-flash, gemini-3-flash-preview, gemini-3.5-flash and gemini-3.6-flash all fail the same way: Plain Text error in stream mode: invalid character ']' looking for beginning of value The OpenAI provider streams fine, so this is the Google provider on the legacy SDK, not the library everywhere. There is a fourth consequence that follows from the third, and it is the nastiest one. LangChainGo installs a streaming function automatically whenever you attach a callbacks handler, in chains/options.go and again in agents/mrkl.go. Since streaming is broken, turning on observability breaks your Gemini agents: Plain Text WITHOUT callbacks handler: err=<nil> out="The current year is 2026." WITH callbacks handler: err=error in stream mode: invalid character ']' looking for beginning of value Feature by Feature CapabilityLangChainGoGenkit GoStructured outputFormat instructions in the prompt, parse the text backgenkit.GenerateData[T], real response schemaToolsCall(ctx, string) (string, error), no schemagenkit.DefineTool[In, Out], typed with JSON schemaAgentsReAct, driven by parsing Action: linesNative function calling, plus an agent primitive with sessions (experimental)CompositionChain interfaces over map[string]anyTyped Go functions (DefineFlow)RetryWrite it yourselfmiddleware.RetryCross-vendor fallbackWrite it yourselfmiddleware.FallbackObservability17 observe-only callback methodsBuilt-in tracing plus a Developer UIOpenTelemetryNoneBuilt inLocal vector storeNone, all 15 stores need a serverlocalvec, file-backedStreaming on GeminiBrokenWorksHTTP endpointnet/http boilerplategenkit.Handler(flow) The single most important row is retry and fallback, and the reason is architectural rather than cosmetic. A Genkit middleware wraps a call and receives a next(), so it can decide to call again, or call something else. LangChainGo's callbacks.Handler returns nothing from every method. You can watch a call, you cannot change it. That missing seam is why every resilience concern ends up in your application code. What We Are Building A support-ticket triage service. Every requirement here is something a real team asks for: Retrieve the relevant policy documents for an incoming ticket.Let the model call a tool to act on it.Return a typed struct, not free text.Retry transient model failures.Survive a vendor outage by falling back to a different provider.Trace every step, so somebody can debug it at 3 am. The Genkit Go Version The whole service, in one file: Go g := genkit.Init(ctx, genkit.WithPlugins( &googlegenai.GoogleAI{}, &openai.OpenAI{}, &middleware.Middleware{}, )) // RAG store. embedder := googlegenai.GoogleAIEmbedder(g, "gemini-embedding-001") store, retriever, err := localvec.DefineRetriever(g, "policy", localvec.Config{Dir: "/tmp/gkvec", Embedder: embedder}, nil) localvec.Index(ctx, docs, store) // A tool the model can call. refund := genkit.DefineTool(g, "issue_refund", "Issues a refund for an order.", func(ctx *ai.ToolContext, a struct { OrderID string `json:"order_id"` AmountCents int `json:"amount_cents"` }) (string, error) { return fmt.Sprintf("refunded %d cents for order %s", a.AmountCents, a.OrderID), nil }) // The flow. triage := genkit.DefineFlow(g, "triage", func(ctx context.Context, ticket string) (*Triage, error) { found, err := genkit.Retrieve(ctx, g, ai.WithRetriever(retriever), ai.WithTextDocs(ticket)) if err != nil { return nil, err } out, _, err := genkit.GenerateData[Triage](ctx, g, ai.WithModel(googlegenai.GoogleAIModelRef("gemini-3.5-flash", nil)), ai.WithDocs(found.Documents...), ai.WithTools(refund), ai.WithPrompt("Triage this ticket using the policy documents, and act on it: %s", ticket), ai.WithUse( &middleware.Retry{MaxRetries: 2}, &middleware.Fallback{Models: []ai.ModelRef{openai.ModelRef("gpt-5-mini", nil)}, ), ) return out, err }) 73 lines, and it returns exactly what you asked for: Plain Text {Category:billing Severity:1 PolicyQuote:Duplicate charges must be refunded in full within 5 business days, no manager approval required. Action:Issued a refund of $25.50 (2550 cents) to resolve the duplicate charge on order A-1029.} The interesting part is which requirements are not code. Retry and fallback are two entries in a list. Tracing does not appear at all, because Genkit instruments flows, generate calls, tool calls, and retrieval itself. The RAG store is three calls because localvec ships in the box. The LangChainGo Version Requirements 1, 2, and 3 port directly. Requirements 4, 5, and 6 do not exist in the library, so they become your code: FileLinesWhy it existsmain.go110the actual applicationmemstore.go68all 15 vector stores need an external service, so the local one is yoursresilience.go58no middleware layer, so retry and fallback are yourstracer.go36observability is a 17-method interface to implementtotal272 None of those three extra files contain interesting code. memstore.go is cosine similarity and a sort. tracer.go is six methods that log and eleven empty ones that exist only to satisfy the interface: Go // Everything below exists only to satisfy the interface. func (t *tracer) HandleText(context.Context, string) {} func (t *tracer) HandleLLMStart(context.Context, []string) {} func (t *tracer) HandleLLMError(context.Context, error) {} func (t *tracer) HandleChainStart(context.Context, map[string]any) {} // ...seven more resilience.go wraps the primary model to add retries and a fallback, and it has a sharp edge worth seeing: Go clean := make([]llms.CallOption, 0, len(opts)) for _, o := range opts { var probe llms.CallOptions o(&probe) // Model names are vendor-specific, and gpt-5-mini rejects 'stop' // outright, which the ReAct agent always sets. Both must go. if probe.Model != "" || len(probe.StopWords) > 0 { continue } clean = append(clean, o) } return r.backup.GenerateContent(ctx, msgs, clean...) CallOption is an opaque function, so to find out what an option does you have to apply it to a probe struct and inspect the result. Miss the stop words and the fallback fails with Unsupported parameter: 'stop' is not supported with this model. Both versions work and produce equivalent results. So far this is a story about volume, not capability. The outage test is where that changes. What Happens When a Model Goes Down Point the primary at a retired model in both versions, which is a real 404 from the live API. Genkit heals and finishes: Plain Text WARN model call failed, falling back model=openai/gpt-5-mini error="Error 404 ... NOT_FOUND" WARN model call failed, falling back model=openai/gpt-5-mini error="Error 404 ... NOT_FOUND" {Category:billing Severity:2 PolicyQuote:Duplicate charges must be refunded in full within 5 business days, no manager approval required. Action:Issued full refund of $25.50 ...} Two calls fail over, once for the initial generate and once for the turn after the tool call. The typed struct survives a vendor switch. LangChainGo heals at the model layer, then breaks one layer up: Plain Text 2026/09/11 20:01:45 primary failed (googleapi: Error 404 ...), falling back 2026/09/11 20:01:54 PARSE FAILED: input text should start with ```json and end with ``` The hand-written fallback did its job. The output parser then rejected the answer because outputparser.Defined.Parse hard-requires code fences: Go const opening = "```json" const closing = "```" if text[:len(opening)] != opening || text[len(text)-len(closing):] != closing { return target, fmt.Errorf("input text should start with %s and end with %s", opening, closing) } Gemini tends to emit those fences. OpenAI does not. The typed output layer is quietly coupled to one vendor's formatting habit, and the coupling surfaces at the exact moment you switch vendors, which is the worst possible time. That same function has a second problem. When the response is shorter than seven characters, it does not return an error; it panics: Plain Text input "ok" PANIC: runtime error: slice bounds out of range [:7] with length 2 input "" PANIC: runtime error: slice bounds out of range [:7] with length 0 An empty completion or a refusal becomes an uncaught panic in your request handler. Dependency Weight The same typed-extraction program, each in its own clean module: LangChainGoGenkit GoModules in the build graph6241Binary size38.0 MB28.7 MB The LangChainGo-only half of that list is worth reading: the deprecated generative-ai-go, plus cloud.google.com/go/vertexai and aiplatform, plus sprig, gonja (a Jinja2 template engine), go.starlark.net (the calculator tool evaluates expressions in Starlark), logrus, tiktoken-go, pkg/errors and json-iterator. A Python-shaped dependency tree, in Go. Where Genkit Go Shines Pulling the thread through all of it, the difference comes down to four things Genkit has, and LangChainGo does not. Middleware is the first. Because a Genkit middleware wraps a call and can decide to call it again or call something else, resilience becomes library code. Retry, cross-vendor fallback, and tool approval are list entries instead of wrapper types you maintain forever. Types are the second. Typed flows, typed tools and schema-backed output mean the compiler carries the contract. That is not just tidier: it is why the Genkit service survived the failover and the LangChainGo one did not. Nothing in the Genkit path was slicing strings out of a model response to find the data. Plugins are the third. A model is a string reference in a registry, so swapping Gemini for GPT is a configuration change. In LangChainGo, two providers are two concrete structs of two different types, which is why a hand-written fallback has to strip vendor-specific call options and still gets it wrong the first time. The Developer UI is the fourth, and it is the one you feel every day rather than during an incident. LangChainGo has no equivalent to it at all. One command starts it next to your app: Shell genkit start -- go run . It reads the registry your code built, so nothing has to be declared twice. On the triage service, it found the flow, the tool, 39 models, and 3 embedders without any configuration: Plain Text Flows (1) triageTicket Tools (1) issue_refund Models (39) googleai/gemini-2.5-flash, ... Embedders (3) googleai/gemini-embedding-001, ... Open a flow, and the input form is generated from your Go struct. A Ticket with Subject and Body fields gives you this, pre-filled and ready to edit: JSON { "body": "", "subject": "" } Press Run and you get the typed output back as a tree, not a blob of text: JSON { "category": "Billing", "urgency": 4, "action": "Refund issued" } The part that earns its place is the trace underneath. Every run is a span tree with per-span timings, and the tool call is a span of its own: Plain Text triageTicket 4.64s ├─ generate 2.52s │ ├─ googleai/gemini-2.5-flash 1.67s │ ├─ issue_refund 1ms │ └─ generate (2) 846ms │ └─ googleai/gemini-2.5-flash 844ms └─ generate 2.12s └─ googleai/gemini-2.5-flash 2.12s You can read the whole shape of the run off that: the first generate called the tool, the model went back for a second turn with the tool result, and a separate generate did the typed extraction. Nobody wrote a line of logging for it. Failures land in the same place. A run that hit a real Gemini constraint showed the error inline with a full Go stack trace, and marked every span in the tree that was affected: Plain Text Error 400, Message: Function calling with a response mime type: 'application/json' is unsupported, Status: INVALID_ARGUMENT, Details: [] That error is worth knowing on its own, and it is model-specific rather than a Genkit limitation. Running the same tools-plus-typed-output call across five models makes the line clear: Plain Text gemini-2.5-flash FAIL Function calling with a response mime type: 'application/json' is unsupported gemini-3-flash-preview OK gemini-3.5-flash OK gemini-3.6-flash OK gemini-3.8-flash OK Gemini 2.5 refuses function calling combined with a JSON response mime type, so on that model a tool call and a typed extraction have to be two steps. Every Gemini 3 model lifted the restriction, which is why the triage service above runs both in one call on gemini-3.5-flash. The useful part is that the Developer UI told me which span failed and why, instead of leaving me to bisect it. The sidebar also carries Traces, Evaluations and Datasets, so the same UI covers running an action by hand, reading what it did, and building a dataset out of it. In LangChainGo, the equivalent of all of this is a callbacks.Handler you write yourself, which returns nothing from every method and, as shown above, breaks your Gemini agents when you attach it. Underneath all four, someone is shipping. When gemini-2.0-flash was retired, that was a 404 to work around in LangChainGo and a fallback that healed itself in Genkit. What LangChainGo Still Does Well This is not a demolition. There are real reasons it is still in use: Provider catalog. Seventeen model providers, including Bedrock, Cloudflare, Ernie, HuggingFace, Maritaca, Mistral, watsonx, and llamafile. Genkit's plugin set is narrower.Vector store catalog. Fifteen integrations, including Dolt, MariaDB, and OpenSearch, which Genkit has no equivalent for.Conversation memory is less typing. chains.NewConversation(llm, memory.NewConversationBuffer()) beats passing resp.History() around by hand. Genkit's answer is an agent session store, which is more capable but sits behind an experimental flag, so on a plain generate call LangChainGo is still terser.The ReAct agent works. On a sequential two-tool task, it succeeds consistently. The cost is the code you own, not flakiness.Familiarity. If your team already thinks in chains, memories, and retrievers from Python, the vocabulary transfers directly. If you depend on a provider or a vector store only LangChainGo supports, that is a real reason to stay. Weigh it against an unmaintained dependency rather than against line counts. The Scorecard Non-blank, non-comment lines. RequirementGenkit GoLangChainGoFull service: RAG + tool + typed output + retry + fallback + tracing73, one file272, four filesBehavior under a vendor outageheals, typed output intactheals at model layer, output parser failsTool with two typed arguments3155 (hand-written unmarshalling)Retry and cross-vendor fallback2 config lines58 lines, vendor-specific options to stripLocal vector store for RAGships as localvec68 lines, no local option existsObservabilitybuilt-in, zero lines36 lines, and it breaks Gemini agentsStreaming on current Gemini modelsworksbroken on all four testedQuickstart with no optionsworks404, default model retiredModules/binary for the same program41 / 28.7 MB62 / 38.0 MB Conclusion If you are starting a Go AI project today, use Genkit Go. That part is not close. If you already have LangChainGo in production, the decision is about risk rather than elegance. The library is eight months without a commit, its Google provider is built on an SDK that reached end of life in November 2025, its quickstart returns a 404, its streaming is broken on every current Gemini model, and turning on observability breaks the agents it is meant to observe. Those are not problems that resolve themselves. The clearest signal is the outage test. Both versions were supposed to survive a vendor failure, and technically both did at the model layer. Only one came back with the data the caller asked for, because in the other one the layer above was matching on code fences. Further reading: How to migrate from LangChainGo to Genkit Go, the step-by-step guideGenkit Go vs ADK Go 2.0Building Agents in Go: Where Genkit Shines vs ADK 2.0Stop using Python for GenAI, use Genkit GoGenkit documentationLangChainGo repositoryMastering Genkit: Go Edition

By Xavier Portilla Edo

CORE

ScyllaDB in Enterprise Java: High-Performance NoSQL With Eclipse JNoSQL

Modern enterprise applications must process increasing data volumes while guaranteeing low latency, high availability, and predictable performance. While relational databases are essential for transactional workloads and complex queries, some applications do not fit the relational model. Systems that manage millions of user engagements, financial events, or IoT data often need a persistence solution built for distributed environments and horizontal scalability. In these cases, ScyllaDB affords a compelling option for enterprise architects. ScyllaDB is a high-performance, distributed NoSQL database based on the wide-column data model. Inspired by Apache Cassandra and compatible with its query language (CQL), ScyllaDB uses a shard-per-core architecture to maximize hardware efficiency and preserve consistent performance under intensive workloads. When should ScyllaDB be chosen over traditional relational databases or other NoSQL options? This article examines its column-family model, compares it with Cassandra, and shows how to integrate ScyllaDB into a Java enterprise application. Modern enterprise applications require more than basic data storage. They need high throughput, low latency, and horizontal scalability to manage complex workloads. ScyllaDB is a distributed, high-performance NoSQL database based on the wide-column data model, designed for applications that require predictable performance at scale. Companies such as Discord, Expedia, and Comcast use ScyllaDB in real-world scenarios, including messaging platforms, real-time applications, and high-volume data processing. ScyllaDB is especially well-suited for event-driven architectures, IoT platforms, financial systems, and applications with demanding read-and-write workloads. ScyllaDB uses a schema-based approach with Cassandra Query Language (CQL), which features SQL-like syntax, including statements such as CREATE TABLE, INSERT, and SELECT. Beyond primitive types, ScyllaDB supports collections like List, Set, and Map, as well as User-Defined Types (UDTs) for modeling complex data. This flexibility involves trade-offs: unlike relational databases, ScyllaDB uses query-driven data modeling, limits queries on arbitrary fields, and requires careful use of secondary indexes. This article will examine ScyllaDB’s column-family architecture, compare it with Apache Cassandra, review its capabilities and limitations, and show how to integrate ScyllaDB into enterprise Java applications. Setting Up ScyllaDB With Eclipse JNoSQL ScyllaDB supports deployment in cloud environments, on-premises data centers, and local development setups. Teams can select either self-managed or managed cloud services based on their infrastructure needs. In this tutorial, we will use a single ScyllaDB instance with Docker to focus on the data model and Java integration, avoiding the complexity of a distributed cluster. To start ScyllaDB locally, execute the following command: Shell docker run -d --name scylladb-instance -p 9042:9042 scylladb/scylla This command launches a ScyllaDB container and exposes port 9042, the default for Cassandra Query Language (CQL) connections. After initialization, the database is ready to accept application requests. Integrating ScyllaDB With Eclipse JNoSQL Next, determine how the Java application will connect to ScyllaDB. Starting with Eclipse JNoSQL 1.1.19, a dedicated driver provides native ScyllaDB support, enabling standard SQL operations alongside ScyllaDB-specific features. In this example, we use a Java application with Eclipse MicroProfile Config, CDI, and Jakarta JSON Processing. These technologies are common in MicroProfile and Jakarta EE environments, which helps with integration. First, add the ScyllaDB dependency to your Maven project: XML <dependency> <groupId>org.eclipse.jnosql.databases</groupId> <artifactId>jnosql-scylladb</artifactId> <version>${jnosql.version}</version> </dependency> Set the ${jnosql.version} property to Eclipse JNoSQL 1.1.19 or a compatible newer version. Configuring the Database Schema ScyllaDB uses a schema-based approach similar to Apache Cassandra. Developers define keyspaces, tables, column types, and user-defined types explicitly with CQL. In this example, we create a keyspace named developers, two tables for people and movies, and several user-defined types. Eclipse JNoSQL supports providing initialization queries through MicroProfile Config. Add the following properties to META-INF/microprofile-config.properties: Properties files jnosql.column.database=developers jnosql.scylladb.query.1=CREATE KEYSPACE IF NOT EXISTS developers WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 1}; jnosql.scylladb.query.2=CREATE COLUMNFAMILY IF NOT EXISTS developers.Person (id bigint PRIMARY KEY, name text, phones list<text>); jnosql.scylladb.query.3=CREATE TYPE IF NOT EXISTS developers.director (name text, movies set<text>); jnosql.scylladb.query.4=CREATE COLUMNFAMILY IF NOT EXISTS developers.Movie (name text PRIMARY KEY, age int, director FROZEN<director>); jnosql.scylladb.query.5=CREATE INDEX IF NOT EXISTS ageIndex ON developers.Movie(age); jnosql.scylladb.query.6=CREATE TYPE IF NOT EXISTS developers.money (currency text, amount decimal); jnosql.scylladb.query.7=CREATE TYPE IF NOT EXISTS developers.headquarter (city text, country text); This configuration highlights multiple key points. CQL is similar to SQL, using statements likeCREATE, explicit types, and primary keys. ScyllaDB supports collections such as list and set, allowing you to represent multiple values without extra tables. User-defined types (UDTs) group related attributes into structured objects, such as the director type. The Movie table uses FROZEN<director>. In CQL, FROZEN stores the UDT as a single value, preventing independent modification of individual fields. We also created a secondary index on the age column. Although ScyllaDB supports secondary indexes, their effectiveness depends on the workload, cardinality, and query types. Do not treat them as equivalent to relational database indexes or as a substitute for proper partition-key design. Defining database schemas through application configuration works for tutorials but isn't recommended for production. Manage schema evolution separately using version-controlled migration tools such as Liquibase, which supports ScyllaDB via its Cassandra extension. This approach separates application deployment from schema changes and improves control over database updates. Mapping Entities and User-Defined Types With the database schema defined, we can create the Java models. We start with Movie, representing a ScyllaDB entity, and Director, representing a user-defined type. Java @Entity("Movie") public class Movie { @Id("name") private String name; @Column private Integer age; @Column(udt = "director") private Director director; } public class Director { @Column private String name; @Column private Set<String> movies; } The Movie entity uses the standard Jakarta NoSQL mapping approach with @Entity, @Id, and @Column. The director attribute maps to a ScyllaDB user-defined type via @Column(udt = "director"). During serialization and deserialization, Eclipse JNoSQL uses the ScyllaDB driver to convert between the Java object and its CQL UDT representation. A UDT is similar to a structured JSON object that groups related fields, but it's a strongly typed CQL value defined in the schema. The Set<String> field in Director demonstrates ScyllaDB’s native support for collections, allowing you to model a director’s movie titles without a separate relational table. Next, we will define the Person entity: Java @Entity("Person") public class Person { @Id("id") private long id; @Column private String name; @Column private List<String> phones; } The phones attribute demonstrates another collection type ScyllaDB supports: list<text>. Eclipse JNoSQL maps Java List<String> directly to the corresponding CQL collection without extra converters or custom serialization. These examples show how ScyllaDB combines a schema-based model with support for structured types and collections. Eclipse JNoSQL offers familiar mapping annotations, permitting applications to use ScyllaDB’s native features. However, object mapping convenience does not replace the need for query-driven data modeling. In distributed databases like ScyllaDB, partition keys, access patterns, and data distribution remain essential architectural considerations. Running the Application: Jakarta NoSQL and Jakarta Data Once you configure the database and map entities, the next step is to interact with ScyllaDB from Java. Eclipse JNoSQL provides two complementary approaches: Jakarta NoSQL, a template-based API for NoSQL databases, and Jakarta Data, which uses the Repository pattern. Both integrate with CDI, letting developers inject persistence components into applications running on either Java SE with CDI or Jakarta EE application servers. Jakarta NoSQL: The Template Approach The first approach uses Jakarta NoSQL with the ColumnTemplate interface, offering a consistent API for wide-column databases. Developers can insert, update, delete, and retrieve entities without interacting directly with the database driver. Java @Inject ColumnTemplate template; For example, you can create a Person entity, insert it, and retrieve it using a fluent query API: Java Person person = Person.builder() .withId(1L) .withName("Ada Lovelace") .withPhones(List.of("234", "432")) .build(); template.insert(person); Optional<Person> result = template.select(Person.class) .where("id").eq(1L) .singleResult(); The entity mapping automatically handles the Java List<String> and its corresponding CQL collection, so you don't need extra serialization code to persist phone numbers. The Template approach is useful when developers need explicit control over persistence operations, want to build queries programmatically, or prefer a consistent API across NoSQL implementations. Jakarta Data: The Repository Approach The second approach uses Jakarta Data and applies the Repository pattern to simplify persistence. Developers define an interface, and Eclipse JNoSQL supplies the implementation. Consider our PersonRepository: Java @Repository public interface PersonRepository extends CrudRepository<Person, Long> { } This interface provides standard CRUD operations, such as save, findById, and deleteById. The repository can be injected with CDI and used directly in the application: Java @Inject PersonRepository repository; repository.save(person); Optional<Person> result = repository.findById(1L); Jakarta Data also supports query methods derived from method names, allowing developers to express queries without writing implementations. MovieRepository demonstrates this capability and exposes ScyllaDB-specific functionality through its dedicated interface: Java @Repository public interface MovieRepository extends ScyllaDBRepository<Movie, String> { List<Movie> findByAge(Integer age); @CQL("SELECT * FROM developers.Movie") List<Movie> findAllQuery(); } The findByAge method demonstrates query derivation, while the @CQL annotation lets you execute native Cassandra Query Language statements. In this example, the age column has a secondary index, so you can query movies by age without using the partition key. However, query derivation does not remove ScyllaDB’s querying restrictions. Repository methods must match queries supported by the schema and indexes. Retrieving all records is useful for demonstrations but should be used cautiously in production with large datasets. Beyond Standard CRUD Operations While Jakarta NoSQL and Jakarta Data simplify common persistence operations, ScyllaDB integration extends beyond standard repositories and templates. Eclipse JNoSQL offers ScyllaDBTemplate, a specialized implementation that exposes database-specific features while maintaining the familiar Java programming model. With this integration, developers can: Control consistency levels: Specify Cassandra-compatible consistency levels, such as ConsistencyLevel.ONE, to balance consistency, latency, and availability.Execute native CQL: Run database-specific queries directly when standard abstractions are insufficient, including those using ScyllaDB-specific features.Work with structured data: Map collections such as List and Set, as well as user-defined types, without manually converting Java objects to CQL values.Choose the right abstraction: Use the vendor-neutral ColumnTemplate for general persistence, or ScyllaDBTemplate for additional database control. For instance, the specialized template supports an operation such as: Java scyllaDBTemplate.save(person, ConsistencyLevel.ONE); A key architectural feature of Eclipse JNoSQL is that abstraction does not mean losing access to database-specific capabilities. Developers can use standardized APIs for common operations and still access native functionality when needed. Jakarta NoSQL and Jakarta Data complement each other, offering different ways to organize persistence logic. The Template approach provides explicit, flexible interaction, while the Repository approach emphasizes declarative data access and separation of concerns. Both benefit from CDI integration, allowing you to use the same persistence components in standalone Java applications, microservices, or Jakarta EE environments. The choice depends on the application’s architectural requirements, but the underlying ScyllaDB capabilities and data-modeling constraints remain the same. Conclusion ScyllaDB is a strong choice for enterprise architects, especially for applications that demand high throughput, consistent latency, and horizontal scalability. Its schema-based wide-column model, support for collections and user-defined types, and compatibility with Cassandra Query Language provide a familiar and robust solution for distributed data persistence. However, to make informed architectural decisions, it's important to understand its query limitations, secondary indexes, and data-modeling trade-offs. Source: https://github.com/JNOSQL/demos-se#scylla.

By Otavio Santana

CORE

Adding Graph Retrieval to a RAG Pipeline With Python

The majority of retrieval-augmented generation (RAG) examples start as follows: document segmentation; create embedding representations for segmented document sections ("chunks"); store those representations in a vector database and use them to "retrieve" the "best matching" document section(s) for any user-inputted search query. That is a good starting point. However, there is one significant limitation to the process described above: not all questions relate solely to similarity. Some questions relate to relationships. For example: Which team owns the API that Checkout Service uses? While a vector search system could potentially identify documents related to "Checkout Service," "Payment API," and "Platform Team," the critical piece of information is the relationship between these concepts: Markdown Checkout Service --DEPENDS ON--> Payment API Platform Team --OWNS--> Payment API This is when graph retrieval becomes a value-added element. Unlike retrieving only documents based on similarity to a query, a graph retrieval layer can retrieve connected Facts from a graph, providing relationship-aware context to a large language model (LLM) while allowing the LLM to generate the final answer. In other words, instead of simply providing additional context to an LLM in terms of a specific set of retrieved documents, a graph retrieval layer would be able to provide structured context. In this tutorial, we will develop a basic graph retrieval layer to enhance a RAG pipeline utilizing Python and NetworkX. We will load relationship triples from a JSON File, build a directed graph, retrieve the local neighborhood of any entity matched by the retrieval layer, and convert that graph-based context into a format suitable for an LLM Prompt. Why Graph Retrieval Helps Most traditional RAG methods are effective for questions that have answers composed within a couple of data "chunks." Examples include: What database does Payment API utilize? As an example, if you have documentation that states, "The Payment API utilizes PostgreSQL," then a vector search should normally identify it. However, as soon as you are looking for information about related facts (i.e., multiple connected pieces of fact), plain similarity search may not provide sufficient results: Who owns the API that the Checkout Service relies upon? In order to determine who owns the API that the Checkout Service relies upon, you need to perform at least two link traversals: Checkout Service -> Payment APIPlatform Team -> Payment API Since this creates a relationship map (a graph), systems, teams, APIs, and databases all represent nodes in the graph, while their relationships all represent the edges in the graph. As long as this structure is in place, retrieval can follow edges (relationships) rather than simply rely on semantic similarities. This does not mean that graph retrieval replaces vector search completely. A hybrid approach has been found to be more successful in most production environments; specifically, the vector search identifies relevant text, while the graph retrieval identifies relevant relationships among those texts, and ultimately the final answer will be determined using both. Therefore, our main emphasis in this article will be solely on the graph retrieval layer. What We Are Building We are creating a short Python application: Loading data (relationship triples) from file "triples.json".Building a directional graph using NetworkX.Identifying all entities mentioned within the user's input query.Retrieving related graph information.Converting that related information to form context for the LLM prompt.Writing the processed contextual information to "graph_context.json." Output will appear as follows: JSON { "question": "which team is responsible for owning the api Checkout Service depends on?", "context": [ "Checkout Service depends_on payment api.", "Platform Team OWNS Payment API." ] } Then you could pass that created contextual information into an LLM prompt portion of a RAG prompt. Project Setup Create a new project folder: Shell mkdir graph-retrieval-rag cd graph-retrieval-rag Create a requirements.txt file: Plain Text networkx==3.3 Install the dependency: Shell pip install -r requirements.txt Create a file named 'triples.json': JSON [ { "subject": "Checkout Service", "relation": "DEPENDS_ON", "object": "Payment API", "source_text": "Checkout Service depends on Payment API during payment authorization." }, { "subject": "Payment API", "relation": "USES", "object": "PostgreSQL", "source_text": "Payment API uses PostgreSQL for transaction metadata." }, { "subject": "Platform Team", "relation": "OWNS", "object": "Payment API", "source_text": "Platform Team owns Payment API." }, { "subject": "Recommendation Service", "relation": "CALLS", "object": "Catalog API", "source_text": "Recommendation Service calls Catalog API to retrieve product details." }, { "subject": "Catalog API", "relation": "USES", "object": "Elasticsearch", "source_text": "Catalog API uses Elasticsearch for product search." }, { "subject": "Search Team", "relation": "OWNS", "object": "Catalog API", "source_text": "Search Team owns Catalog API." } ] This file is intentionally small. In a real system, these triples could come from entity extraction, service catalogs, architecture documents, source code metadata, or cloud inventory. Step 1: Load Relationship Triples Create a file named graph_retriever.py. Python import json def load_triples(path): with open(path, "r", encoding="utf-8") as file: return json.load(file) Each triple has four fields: Plain Text subject relation object source_text The source_text field matters because graph context should remain traceable. When the model uses a retrieved relationship, the system should know where that relationship came from. Step 2: Build a Directed Graph Next, we will create a graph from these triples by importing the NetworkX library for Python, which can create different graph types (for example, directed or undirected). Python import networkx as nx def build_graph(triples): graph = nx.DiGraph() for triple in triples: subject = triple["subject"] object_ = triple["object"] relation = triple["relation"] graph.add_node(subject) graph.add_node(object_) graph.add_edge( subject, object_, relation=relation, source_text=triple["source_text"], ) return graph A directed graph is important because relationship direction changes meaning. This: Plain Text Checkout Service --DEPENDS_ON--> Payment API does not mean the same thing as: Plain Text Payment API --DEPENDS_ON--> Checkout Service Direction matters for dependency analysis, ownership lookup, impact analysis, and many GraphRAG workflows. Step 3: Find Query Entities We're going to find entities mentioned in a question using simple string searching as the means of finding them. Python def find_query_entities(graph, question): question_lower = question.lower() matches = [] for node in graph.nodes: if node.lower() in question_lower: matches.append(node) return matches The code here does not represent a full NLP solution. This is simply a straightforward baseline. In production, you would probably want to do things like aliasing, entity linking, fuzzy matching, etc., or train your own model. In order to normalize the way that a question is phrased versus how something is named in the graph. For instance, "checkout" might be used to ask about the "Checkout Service. Step 4: Retrieve a Graph Neighborhood After identifying the graph element that will answer a question, identify all other elements connected to it. One method to do this would be to gather together both outgoing and incoming edges at each identified node. This could be done with the following code snippet: Python def retrieve_neighborhood(graph, seed_entities, max_hops=2): visited_nodes = set(seed_entities) frontier = set(seed_entities) retrieved_edges = [] for _ in range(max_hops): next_frontier = set() for node in frontier: for neighbor in graph.successors(node): edge_data = graph.get_edge_data(node, neighbor) retrieved_edges.append((node, edge_data["relation"], neighbor, edge_data)) next_frontier.add(neighbor) for neighbor in graph.predecessors(node): edge_data = graph.get_edge_data(neighbor, node) retrieved_edges.append((neighbor, edge_data["relation"], node, edge_data)) next_frontier.add(neighbor) frontier = next_frontier - visited_nodes visited_nodes.update(next_frontier) return deduplicate_edges(retrieved_edges) # Adding a helper function to remove duplicate edges: def deduplicate_edges(edges): seen = set() unique_edges = [] for source, relation, target, edge_data in edges: key = (source, relation, target) if key not in seen: seen.add(key) unique_edges.append((source, relation, target, edge_data)) return unique_edges The max_hops parameter controls how far retrieval travels from the original query entity. For example: Plain Text max_hops=1 retrieves immediate neighbors. max_hops=2 retrieves neighbors of neighbors. While higher values give you a larger amount of additional knowledge about your subject matter, this may also introduce a lot of "noise" into your results. Step 5: Convert Graph Results Into LLM Context The retrieved graph edges are structured data. Before sending them to an LLM, convert them into short, readable facts. Python def format_edges_as_context(edges): context = [] for source, relation, target, edge_data in edges: context.append( { "fact": f"{source} {relation} {target}.", "source_text": edge_data["source_text"], } ) return context This gives the prompt two useful forms of evidence: A normalized graph fact.The original source sentence. That makes the final answer easier to debug and easier to cite. Step 6: Build a RAG Prompt Now create a simple prompt builder. Python def build_prompt(question, graph_context): context_lines = [] for item in graph_context: context_lines.append(f"- {item['fact']} Source: {item['source_text']}") context_text = "\n".join(context_lines) return f""" You are answering a question using graph-retrieved context. Use only the facts provided in the context. If the context does not contain enough information, say so. Question: {question} Context: {context_text} Answer: """.strip() This prompt is intentionally restrictive. The model should use the source instead of guessing. A good production application would also include a format for citations, rules for how to write responses, and fallback behavior. Complete Example Here is the full code: Python import json import networkx as nx def load_triples(path): with open(path, "r", encoding="utf-8") as file: return json.load(file) def build_graph(triples): graph = nx.DiGraph() for triple in triples: subject = triple["subject"] object_ = triple["object"] relation = triple["relation"] graph.add_node(subject) graph.add_node(object_) graph.add_edge( subject, object_, relation=relation, source_text=triple["source_text"], ) return graph def find_query_entities(graph, question): question_lower = question.lower() matches = [] for node in graph.nodes: if node.lower() in question_lower: matches.append(node) return matches def retrieve_neighborhood(graph, seed_entities, max_hops=2): visited_nodes = set(seed_entities) frontier = set(seed_entities) retrieved_edges = [] for _ in range(max_hops): next_frontier = set() for node in frontier: for neighbor in graph.successors(node): edge_data = graph.get_edge_data(node, neighbor) retrieved_edges.append((node, edge_data["relation"], neighbor, edge_data)) next_frontier.add(neighbor) for neighbor in graph.predecessors(node): edge_data = graph.get_edge_data(neighbor, node) retrieved_edges.append((neighbor, edge_data["relation"], node, edge_data)) next_frontier.add(neighbor) frontier = next_frontier - visited_nodes visited_nodes.update(next_frontier) return deduplicate_edges(retrieved_edges) def deduplicate_edges(edges): seen = set() unique_edges = [] for source, relation, target, edge_data in edges: key = (source, relation, target) if key not in seen: seen.add(key) unique_edges.append((source, relation, target, edge_data)) return unique_edges def format_edges_as_context(edges): context = [] for source, relation, target, edge_data in edges: context.append( { "fact": f"{source} {relation} {target}.", "source_text": edge_data["source_text"], } ) return context def build_prompt(question, graph_context): context_lines = [] for item in graph_context: context_lines.append(f"- {item['fact']} Source: {item['source_text']}") context_text = "\n".join(context_lines) return f""" You are answering a question using graph-retrieved context. Use only the facts provided in the context. If the context does not contain enough information, say so. Question: {question} Context: {context_text} Answer: """.strip() def write_json(path, data): with open(path, "w", encoding="utf-8") as file: json.dump(data, file, indent=2) if __name__ == "__main__": question = "Which team owns the API that Checkout Service depends on?" triples = load_triples("triples.json") graph = build_graph(triples) seed_entities = find_query_entities(graph, question) if not seed_entities: print("No matching graph entities found in the question.") raise SystemExit(0) retrieved_edges = retrieve_neighborhood(graph, seed_entities, max_hops=2) graph_context = format_edges_as_context(retrieved_edges) prompt = build_prompt(question, graph_context) result = { "question": question, "seed_entities": seed_entities, "graph_context": graph_context, "prompt": prompt, } write_json("graph_context.json", result) print("Seed entities:") print(seed_entities) print("\nGraph context:") for item in graph_context: print("-", item["fact"]) print("\nPrompt:") print(prompt) Run it: Python python graph_retriever.py Expected output: Python Seed entities: ['Checkout Service'] Graph context: - Checkout Service DEPENDS_ON Payment API. - Payment API USES PostgreSQL. - Platform Team OWNS Payment API. Prompt: You are answering a question using graph-retrieved context. Use only the facts provided in the context. If the context does not contain enough information, say so. Question: Which team owns the API that Checkout Service depends on? Context: - Checkout Service DEPENDS_ON Payment API. Source: Checkout Service depends on Payment API during payment authorization. - Payment API USES PostgreSQL. Source: Payment API uses PostgreSQL for transaction metadata. - Platform Team OWNS Payment API. Source: Platform Team owns Payment API. Answer: The answer is now straightforward: Plain Text Platform Team owns Payment API, which Checkout Service depends on. The important part is not that an LLM can phrase the answer. The important part is that retrieval found the connected facts needed to support it. What This Adds to a RAG Pipeline A standard RAG pipeline often looks like this: Markdown Documents | Chunking | Embeddings | Vector Search | LLM A graph-enhanced pipeline adds another retrieval path: Markdown Question | Query entity detection | Graph neighborhood retrieval | Graph context formatting | LLM In a hybrid system, both paths can work together: Markdown Question | |---- Vector search -> relevant text chunks | |---- Graph search -> connected facts | Combined context | LLM This is useful because vector search and graph retrieval solve different problems. Vector search is good at finding semantically similar text. Graph retrieval is good at following explicit relationships. Production Considerations The graph retrieval layer used in this example is very rudimentary. A production-quality implementation will have many additional "guardrails." Normalized entities are necessary for performing entity matching. Exact string matches will miss most common aliases (i.e., checkout, checkout-service, etc.); however, production-quality implementations will use canonical IDs/aliases/entity linking. Graph retrieval has limits. While there is no limit to how large a graph can grow in terms of returned context, returning too much context can be detrimental. To limit the amount of context returned, implement boundaries, including but not limited to hop count, edge type, node type, confidence value, and source data freshness. All relationships in the graph are not equal in reliability. For example, a dependency established by a service catalog entry may be a more reliable relationship than a dependency inferred from an older document. Therefore, store your confidence level, timestamp, and other source metadata values along with every edge. Not all relationships are equally reliable. Contextual information, in addition to just what was retrieved from the graph, should also contain evidence that supports or refutes the answer. As a result of providing evidence with contextual information, when someone reviews your answer, they will be able to easily verify the answer, and if something does go wrong during development, it will make it easier to debug. Hybrid approaches are best suited for using graph retrieval. While graph retrieval can help the model understand why certain pieces of information relate to each other, document chunks can provide much more detail about those related pieces. When retrieving both types of information, the model will likely receive more complete contextual information. Evaluation is mandatory. You will need to create realistic test questions and determine if the sub-graph retrieved from the database includes enough factual information for answering those test questions. Never assume that graph retrieval will improve performance unless you measure it. Key Takeaways Graph retrieval adds relational contextual knowledge to a RAG framework. A simple directed graph can answer many complex questions involving multi-step logic that would be difficult to accomplish with plain similarity-based search. One key design decision in developing an effective graph-retrieval system is what determines the success of the retrieval layer: does it identify the correct entities and their associated relationships? Graph retrieval works well when the question involves dependency, ownership, impact assessment, lineage, and/or multi-step reasoning. In addition to graph retrieval, build a production-quality retrieval system that includes source tracking, entity normalization, confidence scoring, and evaluation. Try It Yourself Add these triples to triples.json JSON { "subject": "Billing Worker", "relation": "PUBLISHES_TO", "object": "Kafka", "source_text": "Billing Worker publishes payment events to Kafka." }, { "subject": "Data Platform Team", "relation": "OWNS", "object": "Kafka", "source_text": "Data Platform Team owns Kafka." } Then change the question to: Plain Text Which team owns the system that Billing Worker publishes events to? Your goal is to retrieve: Plain Text Billing Worker PUBLISHES_TO Kafka. Data Platform Team OWNS Kafka. Once that works, try lowering max_hops from 2 to 1 and compare the retrieved context. This is a simple way to see why hop count matters in graph retrieval.

By Sriharsha Makineni

Building a Real Pipeline With senro: Monorepos, Caching, and Triggers

In the introduction to senro, I covered the model: a pipeline is a Go program; it builds an immutable plan, and the engine executes that plan while anything can attach to the event stream. This article is the practical follow-up. It covers the five things you will actually write on your first real pipeline: getting your code into a step, caching a step so it never runs the same work twice, fanning one step out over every module in a monorepo, narrowing that fan-out to only what a change touches, and deciding whether a webhook should start a run at all. Everything below was run against senro v1.4.0. A Pipeline That Builds a Real Repository Here is the shape of a Go CI pipeline in senro. Check the code out into a workspace, vet and test it in parallel, then build: Go // Command ci checks the repo out into a workspace, then vets, tests and // builds it. package main import ( "context" "log" "os" "github.com/xavidop/senro" "github.com/xavidop/senro/artifact" "github.com/xavidop/senro/exec" "github.com/xavidop/senro/retry" ) func main() { ctx := context.Background() repo := os.Getenv("REPO_URL") home := os.Getenv("HOME") p := senro.New("ci") // One directory for this run, mounted by every step that needs the code. src := senro.Workspace("src", senro.Scope(senro.ScopeRun)) verify := p.Workflow("verify") verify.Step("checkout", exec.Command("git", "clone", "--depth", "1", repo, ".")). WorkDir("/src"). Mount(src.At("/src", senro.RW)). Retry(3, retry.OnInfra()) verify.Step("vet", exec.Command("go", "vet", "./...")). Needs("checkout"). WorkDir("/src"). Mount(src.At("/src", senro.RO)). Env("HOME", home) verify.Step("test", exec.Command("go", "test", "./...")). Needs("checkout"). WorkDir("/src"). Mount(src.At("/src", senro.RO)). Env("HOME", home) verify.Step("build", exec.Command("go", "build", "-o", "bin/app", "./cmd/app")). Needs("vet", "test"). WorkDir("/src"). Mount(src.At("/src", senro.RW)). Env("HOME", home). Pure(). Inputs(artifact.Glob("**/*.go"), artifact.File("go.sum")). Outputs(artifact.File("bin/app")). Retry(3, retry.OnInfra()) if err := senro.Run(ctx, p); err != nil { log.Print(err) os.Exit(1) } } Plain Text test stdout | ? senrolab/ci [no test files] test stdout | ? senrolab/cmd/app [no test files] test stdout | ok senrolab/greet 0.360s The graph those Needs calls describe, rendered from the run's own plan.json: Plain Text pipeline: ci │ ├─ wave 1 checkout retry 3/infra │ ├─ wave 2 (2 parallel) test │ vet │ └─ wave 3 build needs vet, test · retry 3/infra Two things in there are not obvious coming from a CI system that hands you a checkout and a shell. Workspaces: How Code Reaches a Step Every step runs in an isolated, empty working directory. Nothing hands you a checkout, because "hand the step a checkout" means four different things across a local process, a container, a pod, and an SSH host. A Workspace is a named, versioned directory that steps mount. senro snapshots it into a content-addressed store when a step that mounts it finishes, and restores it for the next step that needs it. So the code gets into the run the way it gets into any CI runner: the checkout step clones it, and the src workspace carries the result to the other three. Mount modes matter: ModeUse it whensenro.RWThe step writes into the workspace, including its declared Outputssenro.ROThe step only reads In the pipeline above, checkout is RW because it writes the tree, vet and test are RO because they only read it, and build is RW because its output lands there. ws.At(path, mode) always needs a mode, Mount accumulates so one step can mount several things, and two mounts at one path are refused at build time. Workspaces come in three scopes: ScopeWhat you getsenro.ScopeRunThe default. One directory for this run, shared by every step that mounts itsenro.ScopePersistentOne directory on this machine that outlives the run. Requires explicit boundssenro.ScopeStepOne fresh directory per step, shared with nobody Declaring the Environment Build() adds nothing to a step's environment. Not a HOME, not a GOPATH, nothing. The local executor supplies its own PATH and that is all. This is deliberate: two developers on the same commit produce the same plan because nothing about their shells leaks into it, and the environment is one of the eleven components a cache key is computed from. In practice, it means tools that need something get told about it: Go .Env("HOME", home) Env takes one pair per call, and it is not variadic on purpose: Env("A", "1", "B") would compile and mean nothing. Caching a Step With Pure() Pure() is a promise: given these inputs, this command produces these outputs, and nothing else matters. senro hashes the inputs, the command, the environment, the workspaces, the mount shape, and six other things into a key. On a second run with the same key, the step is not executed at all, and its recorded outputs are restored. Running the pipeline above twice: Plain Text RUN ID PIPELINE STATUS STARTED DURATION 20260911T105259-8fc9b32e31 ci succeeded 2026-09-11 19:52:59 1s 20260911T105243-8b696eb1d0 ci succeeded 2026-09-11 19:52:43 2s Plain Text $ senro cache explain --run 20260911T105259-8fc9b32e31 build HIT build key a52a9b6c The event stream records it as a state, not a label: JSON {"type":"cache.hit","step":"build", "payload":{"key":"sha256:a52a9b6c...","from_run":"20260911T105243-8b696eb1d0"} {"type":"step.finished","step":"build","payload":{"state":"cached","cached":true} Touch a source file and the same command tells you exactly why it has to run again: Plain Text MISS build key 0e769472 (previous a52a9b6c) ✗ input_digests: ci/main.go ee7563f3 → 4f7d4315 ✗ input_digests: greet/greet.go 59d2873d → 57930022 ✗ input_digests: probe/main.go added ✗ workspace_digests: src f377ce04 → de2069cc ✓ command, env, secrets, executor_class, platform, mount_shape, step_shape, func_identity, tool_versions, version unchanged One rule about the key is worth memorizing, because it decides how you structure a fan-out: Mount puts the whole mounted workspace into the cache key. Inputs narrows what you declare as sources, but it does not narrow the workspace component. A step that mounts a big shared tree has a key that depends on that entire tree. The practical consequence: when you cache expensive per-unit work, give each unit its own workspace instead of sharing one. Read-Only Mounts Are Enforced A step that writes through an RO mount fails, and says so: Plain Text engine: step "build[unit=services/api]" wrote through its read-only mount of workspace "tree" (c9271303 became 832bb638); a read-only mount that changes makes every cache key computed from it wrong That one is go build dropping a compiled binary next to the source. The fix is go build -o /dev/null ./..., which is what a build check wanted anyway. How the rule is enforced depends on the executor, and it is worth knowing which you are relying on: ExecutorWhat senro.RO doesWhen a write is caughtContainerA real read-only bind mountAt the write. It failsKubernetesreadOnly on the volume mountAt the write. It failsLocalNothing at mount timeRight afterward, when the workspace is found changedSSHNothing on the far sideOn read-back. The write is reported, not carried home On local and SSH, read-only is a request senro checks after the fact rather than a rule the kernel enforces. Keep sensitive input out of any workspace a step could overwrite by mistake. Monorepos: Fan Out With Expand Expand(id, graph) adds one step per unit that a graph discovers. Add a module, get a step, and nobody has to write it. Here is a Go workspace with four modules and a dependency chain three deep: Plain Text libs/log <- libs/config <- services/api libs/log <- services/worker Go verify.Expand("build", gowork.Modules()). MaxParallel(4). Needs("checkout"). Affected(src). Template(func(u senro.Unit) *senro.StepBuilder { // WorkDir is the mount point itself, and go -C moves into the unit. return senro.NewStep(exec.Command("go", "-C", u.Dir, "build", "-o", "/dev/null", "./...")). WorkDir("/tree"). Mount(tree.At("/tree", senro.RO)). Env("HOME", home) }) Plain Text checkout -> succeeded build[unit=libs/log] -> succeeded build[unit=libs/config] -> succeeded build[unit=services/worker] -> succeeded build[unit=services/api] -> succeeded One Expand call, four nodes in the plan: Plain Text pipeline: mono (no change filter) │ ├─ wave 1 checkout │ └─ wave 2 (4 parallel) build[unit=libs/config] build[unit=libs/log] build[unit=services/api] build[unit=services/worker] Template is called once per unit in sorted order and must return a fresh *senro.StepBuilder, built with senro.NewStep. Each unit gives you: On UnitWhat it isu.ID, u.NameThe unit's stable identityu.DirThe unit's directory, relative to the rootu.Base()The last path segment: "web" for "apps/web"u.Sources()Every file under the unit, ready for .Inputs(...) on a Pure() template Child IDs come from the expansion ID and the unit (build[unit=libs/log]), never a name the template picks, so the same repository always builds the same graph and a re-run reconstitutes the same children. Two bounds are worth setting. MaxParallel(n) limits how many children of this expansion run at once. MaxNodes(n) rejects an expansion wider than n at build time, default 500, so a scheduler never discovers mid-run that it has forty thousand sandboxes to hold open. Expansion happens once, at Build(). Every unit is discovered and written into the plan before the first step starts. The Eight Unit Graphs GraphDiscoversglob.Dirs / glob.FilesOne unit per matching directory, or per directory containing a matching filegoworkGo workspace modulescargoRust workspace cratesjsworknpm, pnpm or yarn workspacesmaven, gradleJVM modulespyprojectPython projectsbazelBazel packages glob just matches paths. The others read the ecosystem's own manifests, which is what makes the next section work. Running Only What a Change Affects Affected(src) narrows the fan-out to the units a change reaches: the ones owning a changed file, and everything that depends on them, at any depth. The same pipeline, told that one file changed: Plain Text === changed: services/api/main.go === build[unit=services/api] === changed: libs/config/config.go === build[unit=libs/config] build[unit=services/api] === changed: libs/log/log.go === build[unit=libs/config] build[unit=libs/log] build[unit=services/api] build[unit=services/worker] The third case is the point. services/api does not import libs/log. It imports libs/config, which does. That transitive hop is why reading manifests beats globbing paths, and it is why Affected over a glob graph is refused at build time: a path does not know what imports it, and senro would rather say so than quietly run everything. Unaffected units are not skipped; they are not in the plan at all. That is visible in the plan itself. Same pipeline, same Expand, told that libs/config/config.go changed: Plain Text pipeline: mono (libs/config changed) │ ├─ wave 1 checkout │ └─ wave 2 (2 parallel) build[unit=libs/config] build[unit=services/api] Two nodes instead of four, decided at build time. libs/log and services/worker are not present to be skipped, so a re-run of this recorded plan reconstitutes exactly these two. The change source is pluggable: SourceWhat it ischange.FromTrigger(ev)What the webhook that started the run recordedchange.Paths("a/x.go", ...)A literal list, for your own logic and for testschange.Everything()Every unit runschange.Ignoring(src, "docs/**")The above, minus paths you do not care about change.Paths() with no arguments means nothing changed, not everything. Use change.Everything() for that. They are different answers. Keep the Run Directory Out of the Tree senro writes each run to runs/<id>/ under the working directory. That directory holds the run's workspaces, and a workspace usually holds a checkout of the repository. Unit graphs walk the filesystem for manifests, pruning only .git and node_modules, so a second run will discover the first run's checkout as a second copy of every module: Plain Text build[unit=runs/20260911T105558-08e7daff14/ws/tree/services/api] Pin the run directory somewhere outside the tree you fan out over: Go senro.Run(ctx, p, senro.WithDir("/var/lib/senro/runs/"+runID)) Adding runs/ to .gitignore does not help, because the unit graph is walking the filesystem, not asking git. Triggers: The Pipeline Decides Whether to Run Instead of a CI service deciding whether to invoke your pipeline, the pipeline binary is its own matcher. Hand it a webhook delivery, and it decides whether the event is its business: Go err = senro.Run(ctx, pipeline(ev), senro.WithTrigger(ev, // A push to main reports mode "all", which change.FromTrigger passes // straight through: everything builds there. trigger.OnPush(trigger.Branches("main")), // Everything else is narrowed to what it touched. trigger.OnPush(), trigger.OnPullRequest(trigger.Actions("opened", "synchronize")), )) A push to a topic branch that touched services/api/main.go runs one step. A tag push, which this pipeline does not want, runs nothing and says why: Plain Text $ go run ./examples/monorepo --trigger-event /tmp/ev-tag.json senro: trigger: no trigger matched the event: tag refs/tags/v1.2.3 push(branches=[main]): only answers push events, this was tag push: only answers push events, this was tag pull_request(actions=[opened synchronize]): only answers pull_request events, this was tag exit status 78 Exit code 78 is EX_CONFIG, the conventional "this was not my business". One line per matcher explaining why it declined, which is a better answer than a GitHub Actions on: block that silently did not fire. GitHub, GitLab, Bitbucket and Gitea ship, plus your own provider. And the event file is optional: trigger.FromRequest verifies and parses a delivery straight off the wire, per source, so the pipeline binary can be the HTTP endpoint itself. Because the matcher is a binary, you can test a trigger decision locally against a saved payload before it ever reaches a runner. Handlers: Cleanup and Evidence OnFailure runs its handlers once retries are exhausted and the step still failed. Always runs whatever the outcome. A handler can be a Go function, and it knows what it is cleaning up after: Go func Collect(ctx senro.Ctx, _ ReportParams) error { f, ok := ctx.Failure() if !ok { return errors.New("ship/collect only makes sense as a handler") } fmt.Fprintf(ctx.Stdout(), "%s ended %s (exit %d) on attempt %d\n", f.Step, f.State, f.ExitCode, f.Attempt) fmt.Fprintf(ctx.Stdout(), "log tail: %q\n", f.LogTail) return nil } // in the pipeline: build.Step("package", exec.Command("sh", "-c", "echo packaging; exit 1")). OnFailure(senro.Handler("collect", senro.Func("ship/collect", ReportParams{}))). ContinueOnError() Plain Text package ended failed (exit 1) on attempt 1 log tail: "packaging\n" The log tail arrives with the failure, so a handler that classifies errors never has to open a file. Handlers run in the order listed, and a handler that fails never masks the step's own failure: the recorded cause stays the original step. Use senro.Handler(id, action) rather than a workflow's Step. Passing a real step to OnFailure is rejected at build time, because that step would otherwise run twice. A handler also gets its parent's workspaces read-only at the same paths automatically, so Mount on one is refused. Conclusion Put the pieces together, and the argument is not about syntax. The fan-out is a function called once per unit, so what your build does is testable with go test. The Affected graph reads your ecosystem's manifests, so a change to a shared library rebuilds the things that transitively depend on it and nothing else. The cache tells you which of eleven key components changed when it misses. The trigger decision happens in a binary you can run locally against a saved payload. And the whole thing is a Go module with a go.sum, reviewed like the rest of your code. None of that is impossible in YAML. It is that in YAML each one is a plugin, a marketplace action, a service feature, or four hundred lines of bash, and here they are library calls in a language with a compiler. Next in this series: Genkit flows as senro steps: AI pipelines where every model call is a graph node with its own retry, log, and cache entry.Debugging senro pipelines: replaying a finished run, shelling into a live step, and auditing a cache. senro lives at github.com/xavidop/senro.

By Xavier Portilla Edo

CORE

Delta Lake UniForm to Snowflake: Cross-Platform Table Access Without a Second Copy

Making a Databricks-owned Delta table queryable from Snowflake without copying it is one of those things where every individual piece — Delta Lake UniForm, Unity Catalog, private connectivity — is documented somewhere on its own, but nobody writes up how they actually fit together end to end. If you've been asked to give Snowflake read access to a dataset that Databricks owns, without standing up a nightly export job, this is for you. Why the Two Platforms Don't Share Data Natively Delta Lake and Apache Iceberg are both open table formats, but they track a table's files, schema, and change history differently. Delta Lake is Databricks' native format. Snowflake has strong native support for Iceberg. A table written as Delta Lake isn't something Snowflake can open directly, and a table written as Iceberg by some other engine isn't something Databricks reads as Delta. Historically, the fix for this mismatch was a second copy: export the Databricks table, land it somewhere Snowflake could read, and run a job to keep the copy in sync. That's double the storage, a sync job that needs monitoring, and a copy that's stale the moment the source table changes. None of that gets billed as "vendor lock-in" on any invoice, but that's exactly what it is. What UniForm Changes Delta Lake UniForm, generally available since Delta Lake 3.2 (Databricks Runtime 14.3+), writes Iceberg metadata alongside the Delta transaction log on every commit. Same Parquet files on disk, one additional metadata layer. Enabling it is a table property: SQL ALTER TABLE my_table SET TBLPROPERTIES ( 'delta.universalFormat.enabledFormats' = 'iceberg' ); Once that's set, Databricks keeps writing Delta exactly as before. Snowflake reads the same table as Iceberg, through Unity Catalog's Iceberg REST Catalog endpoint. No export step, no second storage bill. The Setup, End to End Getting from "UniForm is on" to "a Snowflake analyst can query the table" is four pieces. Piece 1: Enable UniForm on the source table. Covered above. One property, per table, Runtime 14.3 or later. Piece 2: Scope a service principal at the Unity Catalog level, not the storage account. Register a service principal and grant it Unity Catalog privileges — EXTERNAL USE SCHEMA specifically, which isn't part of ALL PRIVILEGES and is easy to grant around by accident. There's no separate storage role to assign. Unity Catalog holds the storage credential itself and vends short-lived, scoped credentials to Snowflake at query time, so the service principal only ever authenticates against the catalog API, never against the storage account directly. Piece 3: Put the connection on private networking. This is the piece that involves the most moving parts, and it's worth walking through because the Snowflake side does the provisioning, not Databricks. Snowflake account needs to be on Business Critical edition or higher for this, and both accounts need to be in the same Azure region. Provision an outbound private endpoint from Snowflake toward the Databricks workspace: SQL SELECT SYSTEM$PROVISION_PRIVATELINK_ENDPOINT( '<databricks_resource_id>', 'databricks_ui_api' ); That endpoint sits in a Pending state until someone with access to the Azure resource approves it in the Azure Portal — a step that's easy to lose a day to if you don't know to go looking for it. Then register the catalog integration: SQL CREATE OR REPLACE CATALOG INTEGRATION uc_catalog_int CATALOG_SOURCE = ICEBERG_REST CATALOG_API_TYPE = PRIVATE CATALOG_URI = 'https://<databricks-workspace-url>' ... ENABLED = TRUE; The detail that trips people up: use the standard public hostname for the workspace URL, not the PrivateLink endpoint's own hostname. Snowflake routes traffic through the private endpoint automatically once it exists — pointing the catalog integration at the PrivateLink hostname directly just breaks it. A second private endpoint, and an external volume with USE_PRIVATELINK_ENDPOINT = TRUE, does the same thing for the storage account so file reads stay off the public internet too. Piece 4: Link the catalog — don't hand-create Iceberg tables. The last step is a catalog-linked database, not individual CREATE ICEBERG TABLE statements: SQL CREATE DATABASE uc_linked_db LINKED_CATALOG = ( CATALOG = 'uc_catalog_int' ) EXTERNAL_VOLUME = 'uc_external_vol'; Once this exists, whatever is in Unity Catalog — including every UniForm-generated Iceberg table — shows up in Snowflake automatically, and stays in sync on its own. New tables and schemas created later in Databricks appear without touching Snowflake again. It's a refresh on an interval under the hood, not a live push, so don't oversell the freshness to whoever's asking for "real-time." Writes Are Newer, and Direction Still Matters Everything above is Snowflake reading a Databricks-written table. Writing the other way is newer, and support isn't symmetric yet. Snowflake now supports writing to Databricks-managed Iceberg tables in Unity Catalog — GA on Azure as of earlier this year. Snowflake Horizon's federation is bidirectional by design, using vended short-lived credentials rather than long-lived IAM roles in either direction. Unity Catalog's federation is still mostly inbound: strong at letting other engines read and write into it, weaker when Databricks itself needs to reach out to a catalog it doesn't own. If a design depends on writes going both directions, check current support for the specific cloud and catalog combination before committing to it — this part is moving fast. Where It Still Falls Short Row- and column-level masking is GA for Spark and expanding to other Iceberg REST-compatible engines, but Snowflake's Scan Plan API — cross-engine, attribute-based access control for federated Iceberg data — was still in private preview as of mid-2026. Don't assume a masking policy enforces the same way on both sides until it's been tested. Iceberg's own spec keeps moving underneath this too. Iceberg v3 features like deletion vectors and row lineage are in public preview on Databricks now, which is good news for performance, but it means "Iceberg compatible" isn't a fixed target — check that both platforms agree on the table spec version before standardizing on it. And disaster recovery for federated catalog setups is still gated or preview-stage on the Unity Catalog side as of this writing. Don't lean on this as part of a DR plan without confirming that directly. One Thing I'd Tell Anyone Doing This Check the edition and region requirements before you design anything, not after. Business Critical or higher on the Snowflake side, same Azure region on both sides — if either of those isn't true for your accounts, the private connectivity piece doesn't work at all, and there's no workaround short of moving one of the accounts. Confirm that first, then start building.

By Jeevan Krishna Paruchuri

Rebuilding Elasticsearch Under Load: Unsafe Watermarks and a Sampling Blind Spot

Rebuilding an Elasticsearch index from PostgreSQL sounds simple: Create a new index.Copy the rows that belong in search.Replay anything that changed during the copy.Verify the result.Move the search alias. The difficult part is deciding where the bulk copy ends and the replay begins. A transaction can update a row before the rebuild starts and commit after the copy has taken its database snapshot. With the wrong watermark, that change may appear in neither the copy nor the replay. The rebuild finishes successfully, but the new index is wrong. I built a controlled recovery drill to measure this problem. The test used synthetic data rather than information from a client or production system. The setup included: 500,000 PostgreSQL rows449,917 rows initially matching the search predicate2,000 updates, 250 deletes, and 120 inserts during each rebuildEight concurrent writer connectionsA two-node Elasticsearch 8.19.3 cluster I ran the timestamp-based strategy five times. It produced the same 366 measurable inconsistencies in every run. The corrected transaction-aware strategy was tested in 30 runs across several configurations. All 30 passed the implemented pre-cutover checks. The most surprising result was not the total number of errors. It was that a field-level comparison of 5,000 documents reported no stale content in any of the five defective runs. First, Confirm That a Rebuild Is Necessary Before rebuilding from PostgreSQL, confirm that the existing index cannot be recovered. Resolve the application alias, identify the physical index behind it, and inspect the affected shards. Elasticsearch’s cluster allocation explain API can show whether an unassigned shard is caused by disk pressure, a missing node, an allocation rule, or the absence of a valid shard copy. If a usable snapshot exists, restoring it into a separate versioned index may also be faster than rebuilding the full projection. The rest of this article assumes that allocation repair and snapshot restoration are not available. Unsafe Watermark 1: updated_at The first rebuild strategy used a timestamp as the replay boundary: SQL -- Record the replay boundary. SELECT clock_timestamp(); -- Copy the current search projection. SELECT id, title, body, status, version, updated_at FROM searchable_content WHERE status = 'PUBLISHED' ORDER BY id; -- Replay rows changed after the boundary. SELECT id, title, body, status, version, updated_at FROM searchable_content WHERE status = 'PUBLISHED' AND updated_at > :rebuild_started_at ORDER BY id; This approach assumes that updated_at represents commit order. In the drill, the value was assigned when the update statement ran, not when the transaction committed. Consider this sequence: Plain Text Session A: UPDATE row 1 updated_at = 17:23:46.252262 transaction remains open Session B: reads the rebuild watermark watermark = 17:23:46.465777 Session A: COMMIT The transaction committed after the watermark, but the row carried a timestamp from before the watermark. The bulk scan did not include the update because its PostgreSQL snapshot was created while the transaction was still open. The replay did not include it because the row failed this condition: SQL updated_at > :rebuild_started_at The change fell between the two phases. This issue is not limited to clock_timestamp(). A timestamp assigned by an application, database statement, or trigger may still be created before the surrounding transaction commits. If application servers assign the value, differences between their clocks introduce another source of error. Timestamp replay also has two simpler problems. First, a deleted row no longer exists in the source table. A query against the current rows cannot tell Elasticsearch that the old document must be removed. Second, a row may remain in PostgreSQL but stop matching the search predicate. In this experiment, a row could change from PUBLISHED to DRAFT. The replay query selected only published rows, so it did not remove the old Elasticsearch document. At the end of each timestamp-based run, the defects were: Defect classCountMissing from Elasticsearch3Deleted from PostgreSQL but still indexed228Outside the search predicate but still indexed131Present in both systems but stale4Total366 PostgreSQL contained 449,863 rows matching the final predicate, while Elasticsearch contained 450,219 documents. The net count difference was 356, which understated the real problem. Three missing documents offset part of the 359 unexpected documents, and the four stale documents did not change the count at all. Unsafe Watermark 2: MAX(seq) A transactional outbox solves part of the problem. A trigger can write an outbox event in the same transaction as every insert, update, or delete. Deletes now leave tombstones, so the replay process knows which Elasticsearch documents to remove. A first outbox design might use a BIGSERIAL sequence. Before starting the copy, the rebuild records the largest visible sequence value: SQL SELECT MAX(seq) FROM reindex_outbox; After the copy, it replays everything above that value: SQL SELECT seq, content_id, op FROM reindex_outbox WHERE seq > :starting_seq ORDER BY seq; This avoids clock problems, but it still does not provide a commit boundary. Consider two transactions: Plain Text T1: UPDATE row 1 outbox seq 51 allocated transaction remains open T2: UPDATE row 2 outbox seq 52 allocated COMMIT Rebuild process: SELECT MAX(seq) returns 52 T1: COMMIT When the rebuild reads MAX(seq), T1 is still invisible. T2 has committed, so the largest visible sequence is 52. T1 later commits with sequence 51. A replay using seq > 52 will never return the T1 event. PostgreSQL sequences generate distinct values, but they do not represent commit order. The PostgreSQL sequence documentation also explains that a value returned by nextval() is not reclaimed if its transaction later aborts. Sequence allocation and transaction visibility are separate mechanisms. Replacing the timestamp with a counter did not fix the underlying problem because both values were assigned before transaction visibility was settled. Figure 1. Timestamp and sequence-maximum watermarks can both miss a transaction that starts before the watermark and commits afterward. A Transaction-Aware Replay Window The corrected implementation records the PostgreSQL transaction ID in every outbox event: SQL CREATE TABLE reindex_outbox ( seq bigserial PRIMARY KEY, content_id bigint NOT NULL, op text NOT NULL CHECK (op IN ('upsert', 'delete')), xact_id bigint NOT NULL DEFAULT pg_current_xact_id()::text::bigint, captured_at timestamptz NOT NULL DEFAULT clock_timestamp() ); The outbox event is inserted in the same transaction as the source change. Before beginning the bulk scan, the rebuild records the current PostgreSQL snapshot’s xmin: SQL SELECT pg_snapshot_xmin( pg_current_snapshot() )::text::bigint; PostgreSQL defines snapshot xmin as the lowest transaction ID that was still active when the snapshot was taken. This gives the rebuild a conservative lower bound for transactions that may not be represented in the bulk-copy snapshot. The replay population begins with: SQL SELECT DISTINCT content_id FROM reindex_outbox WHERE xact_id >= :snapshot_xmin; This may include changes that the bulk scan already copied. That is acceptable because the Elasticsearch operations use stable document IDs and are designed to be idempotent. The snapshot tells the rebuild where replay should begin. It does not tell the rebuild when replay is complete. A Final Boundary Is Still Needed An outbox consumer may catch up while writes continue. It may even report zero lag for a moment. That does not close the recovery window because another transaction can commit immediately afterward. The drill therefore uses a final quiescent boundary: Capture the transaction-aware replay lower bound.Bulk-copy the source projection while writes continue.Drain visible outbox events.Stop accepting new source mutations.Allow transactions already in progress to finish.Drain the remaining events.Verify the new index.Move the alias and resume writes. The synthetic writer had a finite plan and had already completed by the time the final barrier was applied. The experiment therefore measured finalization after the writer became idle. It did not measure the delay that production requests would experience while blocked by an active write barrier. A system that cannot pause writes needs a durable stream handoff. PostgreSQL logical decoding, for example, can combine an exported snapshot with a persistent replication position. Do Not Use One Sequence Cursor for the Drain My first outbox drain repeated the sequence problem in another place. It used: SQL WHERE seq > :last_applied_seq Suppose sequence 52 becomes visible while the transaction holding sequence 51 remains open. If the consumer advances its cursor to 52, sequence 51 is skipped when it later commits. The corrected implementation records applied events individually: SQL SELECT o.seq, o.content_id FROM reindex_outbox o LEFT JOIN reindex_progress p ON p.rebuild_id = :rebuild_id AND p.seq = o.seq WHERE o.xact_id >= :snapshot_xmin AND p.seq IS NULL ORDER BY o.seq; This is an applied-event ledger rather than one high-water mark. A late commit cannot appear below a cursor that has already moved past it. The replay also reads the current PostgreSQL state instead of trusting an old document payload stored in the outbox: If the row currently matches the search predicate, it is upserted.If the row was deleted, the Elasticsearch document is deleted.If the row still exists but no longer matches the predicate, the document is also deleted. Several events for the same source ID therefore converge on its current state. An outbox event is marked as applied only after its Elasticsearch operation succeeds. Unresolved operations remain unapplied and cause the cutover to fail. The Elasticsearch Bulk API also requires item-level checking. A request can return HTTP 200 while individual operations fail and the response contains errors: true. In a separate overload test, 525 rejected document operations appeared inside 22 HTTP-200 responses. Checking only the HTTP status would have reported those requests as successful. Why a 5,000-Document Sample Reported a Clean Result I tested three lightweight reconciliation methods against the timestamp-based rebuild: Counts divided into six ID rangesChecks against the 250 IDs selected for deletionA field-level comparison over 5,000 randomly selected source records The first two checks found problems immediately. All six count ranges differed, and 228 of the 250 deleted IDs were still present in Elasticsearch. The sampled content comparison found none of the four stale documents in any of the five runs. The full-corpus comparison found all four. For a population of N comparable documents containing s stale records, and a sample of n documents selected without replacement, the probability of finding at least one stale record is: Plain Text P(detect at least one) = 1 - C(N - s, n) / C(N, n) For this experiment: Plain Text N = 449,860 documents present in both systems s = 4 stale documents n = 5,000 sampled documents The probability of finding at least one stale document was approximately 4.37%. To reach a 95% detection probability, the sample would need 237,134 documents, or 52.7% of the comparable corpus. Reaching 99% would require 307,601 documents, or 68.4%. Figure 2. Detection probability as the sample size increases when four of 449,860 comparable documents are stale. This does not mean sampling is useless. A 5,000-document sample can quickly reveal a widespread transformation error. Sampling can also be useful as a regular monitoring check between rebuilds. It is weak as a cutover gate for rare defects. Transaction-boundary races may affect only a few records inside an otherwise reasonable-looking index. That is exactly the situation in which a modest random sample has little statistical power. What the Full-Corpus Check Compared The full comparison checked: The complete expected set of document IDsA digest over id, title, body, status, and version Each field was length-prefixed before hashing so that field boundaries could not be confused. PostgreSQL calculated the source-side digest. On the Elasticsearch side, the verifier read the actual fields from _source and calculated the digest again. The drill uses MD5 as a compact equality checksum, not as a security mechanism. The digest excludes updated_at, whose representation was not canonicalized in the same way on both sides. The result was: Reconciliation resultTimestamp rebuildTransaction-aware rebuildMissing from Elasticsearch30Deleted source rows still indexed2280Predicate-excluded rows still indexed1310Stale in 5,000-document sample00Stale in full-corpus comparison40Total measured inconsistencies3660 The digest check verifies the document fields included in this projection. It does not verify mappings, analyzers, synonyms, autocomplete, facets, relevance, or authorization rules. Those require functional search tests before the alias moves. For a much larger index, the comparison should stream sorted ID-and-digest pairs or compare deterministic partitions instead of loading everything into memory. A Practical Pre-Cutover Gate A rebuild should not move the alias until all of these conditions are true: The expected and actual document IDs match.The full-corpus content digests match.No committed outbox event in the recovery window remains unapplied.No Elasticsearch operation remains unresolved.The target index has the required active shards.Representative search canaries pass. The canaries should test the behavior users depend on, not only whether one document exists. Depending on the application, they may include: A known search queryAn autocomplete requestA filter or facetSpecial-character or multilingual contentA tenant or authorization boundary Elasticsearch supports removing the old alias target and adding the new target in one alias update operation. The response should still be checked, and the previous index should be retained until the rollback window has passed. It is also important to measure through verification. In this experiment, final drain, shard preparation, and full-corpus verification added roughly 6.5 to 13.8 seconds after the main rebuild work. A benchmark that stops at the last bulk request does not measure when the index is actually ready. Limitations This was a controlled experiment using synthetic data. The source contained one table, one search predicate, and no joins or enrichment. The main benchmark used one primary shard. Both Elasticsearch nodes ran inside one Podman virtual machine on an Apple M4 Max laptop. This allowed a replica to be allocated, but it did not represent separate machines, storage devices, network paths, or failure domains. The workload plan was seeded, so the operations and target IDs repeated across runs. Thread scheduling, timing, and throughput did not. The writer completed before the final barrier was activated, so the experiment measured quiescent finalization rather than the delay experienced by blocked production writes. The content digest covered five fields and excluded updated_at. It verified document content, not mappings or search behavior. Finally, snapshot xmin is a conservative recovery boundary, not a replacement for a mature change data capture system. Long-running transactions can widen the replay range. Outbox events must be retained for the full rebuild window, and every relevant source mutation path must participate in the same transactional capture process. Conclusion The timestamp watermark failed because the row timestamp was assigned before its transaction committed. MAX(seq) failed for the same reason: sequence allocation did not represent commit order. The reconciliation test exposed a separate weakness. Checking 5,000 documents sounded substantial, but it provided only a 4.37% chance of finding the four stale records created by this workload. A reliable rebuild needs two guarantees: A replay window based on transaction visibility rather than statement timingA complete pre-cutover comparison capable of finding rare inconsistencies Without both, the rebuild can complete, the document count can look reasonable, and the new index can still disagree with PostgreSQL. Companion Materials The synthetic workload, implementation, figures, and per-run results are available in the dzone-recovery-drill-v1.1 companion repository.

By Tarun Sharma

How Kestra Unifies Enterprise Workflow Orchestration and Agentic AI

Kestra is an open-source workflow orchestration platform that unifies IT scheduling, data pipelines, business processes, infrastructure automation, and AI agents under one declarative, event-driven control plane. As enterprises modernize legacy systems and adopt agentic AI, unified enterprise workflow orchestration is becoming increasingly important. Here is why I joined Kestra as Global Field CTO and why I believe the orchestration market is approaching a turning point. Orchestration is the connective tissue of the modern enterprise, and right now nobody owns it. That is the gap I am joining Kestra to help close. The market is at the same inflection point Apache Kafka was at when I joined Confluent in 2017. The technology is mature, already in production at Apple, JPMorgan, Toyota, and Xiaomi. The commercial story is under-narrated. And there is a two-to-three-year window to define who owns the category before it closes. If you want the backstory on the nine years that led here, including the journey from a Kafka startup to an IBM acquisition, I wrote about it separately in My Confluent Chapter: From Apache Kafka Startup to $11 Billion IBM Acquisition. Enterprise Workflow Orchestration: The Problem With Disconnected Tools Every large enterprise runs four kinds of orchestration tools that do not talk to each other. Legacy IT schedulers like Control-M, ActiveBatch, and cron run date-driven batch jobs with no event awareness. Data pipeline tools like Airflow, Prefect, and Dagster move data, but stay scoped to Python DAGs. Business process platforms like Camunda, Appian, Pega, and UiPath handle approvals and human steps, but stop at the process boundary. Infrastructure automation tools like vRA, ServiceNow, and Terraform provision resources, but stay locked into ITSM lifecycles. Each tool owns one slice. None of them owns the full picture. Running these tools separately was manageable when each stayed in its lane. Two forces are changing that: enterprises are actively modernizing these tools, and agentic AI is forcing them to work as one. Why Enterprise Orchestration Matters for Agentic AI The first force is modernization, and it is where most enterprises actually start. They are retiring brittle, expensive legacy schedulers. Consolidating data pipeline sprawl across teams. Replacing rigid BPM suites that stop at the process boundary. Each of those is a funded project on its own. And once you are modernizing one category, the obvious question follows: why run four engines when one declarative platform covers all four? The second force is agentic AI, and it removes any remaining reason to keep the tools apart. AI workflows refuse to stay inside a single tool. One execution can touch a data pipeline, trigger a process step, call a model, and provision infrastructure. When it fails, it often fails quietly in the gap between tools. The CTO sees governance risk. The CIO sees infrastructure that cannot keep up. The CDO sees stale data. The COO sees a process that stalled for no visible reason. Four executives, four symptoms, one root cause: no single layer governs the full mix of workloads. Kestra takes all four categories onto one declarative platform, under one governance model. Instead of a patchwork of disconnected tools, one engine orchestrates across the domains an enterprise depends on. Deterministic workflows and autonomous agents, running side by side, under one control plane. Kestra: Open-Source Orchestration for Cloud, On-Premises, and Edge Kestra is open source under the Apache 2.0 license, the same permissive license as Apache Kafka. Developers and data engineers can start in minutes, building workflows in YAML, through APIs, or with the no-code editor, with a plugin ecosystem of 1,500+ integrations and tasks that run code in any language (Python, Node.js, R, Go, Shell, and more). Workflows can even be generated by an LLM via MCP, so you describe the orchestration in natural language and let the model build the flow. The platform is language-agnostic and event-driven by design. For production, there is an enterprise edition and a fully managed cloud, with the governance, scale, and SLAs that regulated organizations require built in rather than bolted on, including air-gapped deployment for those that cannot let a single byte leave their perimeter. Proven in Production at Apple, JPMorgan, Toyota, Xiaomi, and Many Other Enterprises Across Public and Private Industries Kestra runs in production well beyond early adopters. Apple's machine learning team uses it to orchestrate large-scale data pipelines. A senior engineering manager there put it plainly: "I want to highlight their robustness, which is crucial at our scale." Few orchestration tools earn that kind of statement from a team operating at Apple's level in AI and ML. The enterprise logos span industries and regions: Bloomberg, Crédit Agricole, JPMorgan, Toyota, T-Systems, and Xiaomi among them. Behind those production deployments sits a fast-growing open-source community, with execution volume up roughly 20x year over year. The growth curve is the same shape I watched at Confluent. Community adoption first, enterprise gravity following close behind. Enterprise Workflow Orchestration in the Age of AI I do not join companies for incremental bets. I expect Kestra to be a high-growth startup operating at enterprise scale, with the largest companies in the world adopting it across business units for a widening set of use cases. That is not optimism. It is where the demand is already heading. Open, scalable, reliable, and flexible workflow orchestration is needed everywhere. Every business unit runs processes that have to be triggered, coordinated, retried, audited, and governed. Finance, operations, data, platform engineering, customer-facing teams: each has its own workflows, and today each reaches for a different tool. One platform that serves all of them is a large and durable market, not a niche. Agentic AI raises the stakes further. It does not reduce the need for orchestration; it multiplies it. Every agent that takes an action needs a governed, observable workflow underneath it, and those workflows cut across exactly the business-unit boundaries that today's fragmented orchestration tools cannot span. The more AI an enterprise adopts, the more it needs one control plane across every unit. That is the wave Kestra is built to ride. Kestra Architecture: Database and Apache Kafka Backends Start simple. Run Kestra on a standard database and never think about Kafka at all. For many workloads, that is all you need. The platform is event-driven from the ground up, and the database backend handles low- and medium-scale with no extra moving parts. After nine years close to Apache Kafka, people expect me to lead with the Kafka angle. So let me deal with it directly, because there is a real and underappreciated opportunity here. But it is an option, not a requirement. I have seen too many teams force the wrong tool into orchestration. Some use Kafka Streams or Flink to orchestrate workflows. In most cases, that is an anti-pattern. Neither was built for it. Others bolt on yet another complex system just to get scale. Others hack together Python scripts and hope they hold. Kestra takes a cleaner path. Run it on a traditional database for low and medium scale. Switch to an existing Kafka deployment when you need larger throughput. Light when you want it light, Kafka-native when you need the scale. For the thousands of enterprises that already run Kafka as their nervous system, that is a natural fit. Everyone else can stay on the database. Today that means database or Kafka. With the re-architected core in Kestra 2.0 (coming soon), queue and database become fully independent and pluggable, adding support for Redis, AMQP, and cloud-native messaging from AWS, GCP, and Azure. The orchestration layer adapts to your infrastructure rather than forcing you onto theirs. But Kafka is one entry point, not the destination. The bigger opportunity sits in everything beyond the place I came from: legacy scheduler modernization, data and AI pipelines, and business process orchestration across the enterprise. Why I Joined Kestra as Global Field CTO Global Field CTO is the external technology and business voice of Kestra to customers, partners, analysts, press, and the practitioner community. The job is to help Kestra grow across these orchestration domains, regions, and the community building on it. I have written before about the daily life of a Field CTO, and the shape of the role carries over directly. In practice, that means three things. Working with enterprise CIOs and CTOs to figure out where orchestration fits in their architecture, in briefings and roundtables. Briefing analysts like Gartner, Forrester, and IDC on the customer patterns and market shifts I see. And doing the public work: reference architectures, conference keynotes, competitive positioning, and content that maps the technology to concrete use cases across financial services, insurance, manufacturing, automotive, telco, retail, logistics, and technology. That last part is where I have spent years building signature assets. The landscape reports, the industry guides, the books. I will keep producing them, now with orchestration as the connecting layer across data integration, process intelligence, and agentic AI. I Recommend the Right Architecture, Including When It Is Not Kestra One thing will not change from how I worked before. I recommend the right architecture, even when Kestra is not the answer. Orchestration rarely lives alone. A company might run Celonis for process mining to understand its operations, and Kestra for process orchestration at scale once agentic AI starts acting on those insights. Another might keep Camunda for human-centric, BPMN-modeled approval workflows, and use Kestra for the technical orchestration underneath: the data pipelines, integrations, and agent steps those processes depend on. The two are complementary in both cases. My job is to be clear about where Kestra fits, where it does not, and how it works alongside the rest of the stack. Credibility is what makes this role effective. I intend to keep mine. Europe Today, the US Next Kestra has its deepest enterprise traction in Europe today. The next move is the United States, where the company is expanding strategically over the coming months and opening a New York presence. I will support that push heavily. Customers in Asia and Australia are already in production, and the Middle East is a deliberate next step. Let's Build This Together There is a lot of ground to cover, and a lot of flights ahead to meet customers, partners, and the broader open-source community in person. I am excited to get started. If you are wrestling with orchestration across data, processes, infrastructure, or agentic AI, I want to be in that conversation. The simplest way: pull me into your architecture discussions. Reach out, and let's work through the hard problems together.

By Kai Wähner

CORE

Two Engines, One Query: DuckDB Inside Neo4j

If you've been following the database world lately, you'll have noticed DuckDB turning up in some unexpected places. One of the more interesting examples is pg_duckdb, a Postgres extension that embeds DuckDB directly into the Postgres process, letting analytical queries run through DuckDB's engine without leaving the database. It's a clever idea: keep your data where it is, but bring a faster analytical engine. That got us wondering: could we do something similar with Neo4j? The answer is not in the same way. Postgres has a mature extension architecture that allows foreign query engines to be embedded at a low level. Neo4j doesn't expose that kind of hook. But Neo4j does let you write custom procedures in Java that run inside the Neo4j JVM and are callable directly from Cypher. Also, DuckDB ships an official JDBC driver that bundles its native binary and loads it via JNI. Put those two things together and you've got something that feels surprisingly close to the pg_duckdb experience: DuckDB running in-process, queries fired from Cypher, results flowing back into a graph. That's what we built. In this article, we'll walk through the whole thing from scratch. The full source code is available on GitHub. Prerequisites Before starting, we'll need the following: JDK 21 – check with java -versionMaven – check with mvn -versionNeo4j Desktop – with a local database instance created and ready to start Note that this approach requires a self-managed Neo4j install. AuraDB doesn't support custom plugins, so Neo4j Desktop is the right choice here. Version Summary For reference, here are all the component versions used in this article: ComponentVersionOpenJDK21.0.12.1Maven3.9.6Neo4j Desktop2.2.1Neo4j2026.07.0Neo4j Procedure API5.26.0DuckDB JDBC1.1.3JUnit Jupiter5.10.2maven-compiler-plugin3.11.0maven-shade-plugin3.5.2maven-surefire-plugin3.5.4 Getting the Code The full source code is on GitHub. To get just this project: Shell cd ~ git clone --filter=blob:none --sparse https://github.com/VeryFatBoy/neo4j.git cd neo4j git sparse-checkout set duckdb-procedure mv duckdb-procedure ../duckdb-procedure cd ../duckdb-procedure/neo4j-duckdb-procedure Or follow along from scratch using the steps below. How It Works The architecture is straightforward. We write two Java classes and package them as a fat jar: DuckDBConnection manages a singleton DuckDB connection that lives for the lifetime of the Neo4j process. It uses double-checked locking to keep things thread-safe and reconnects automatically if the connection is ever closed.DuckDBProcedure is the Neo4j procedure itself. It accepts a SQL string from Cypher, runs it against the shared DuckDB connection and streams the results back as rows, each one a Map<String, Object> that Cypher can destructure with dot notation. DuckDB runs in-memory by default, which means it starts clean every time Neo4j restarts. That's fine for analytics against external files, and we'll discuss file-backed persistence in the article Summary section. Project Setup Start by creating the project directory: Shell mkdir -p neo4j-duckdb-procedure/src/main/java/com/example/duckdb cd neo4j-duckdb-procedure Verify the structure: Shell find . -type d Now create pom.xml in the project root. XML <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.example</groupId> <artifactId>neo4j-duckdb-procedure</artifactId> <version>1.0.0</version> <packaging>jar</packaging> <properties> <java.version>21</java.version> <maven.compiler.source>21</maven.compiler.source> <maven.compiler.target>21</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <neo4j.version>5.26.0</neo4j.version> <duckdb.version>1.1.3</duckdb.version> </properties> <dependencies>  <dependency> <groupId>org.neo4j</groupId> <artifactId>neo4j</artifactId> <version>${neo4j.version}</version> <scope>provided</scope> </dependency>  <dependency> <groupId>org.duckdb</groupId> <artifactId>duckdb_jdbc</artifactId> <version>${duckdb.version}</version> </dependency>  <dependency> <groupId>org.junit.jupiter</groupId> <artifactId>junit-jupiter</artifactId> <version>5.10.2</version> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.11.0</version> <configuration> <source>21</source> <target>21</target> </configuration> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-surefire-plugin</artifactId> <version>3.5.4</version> </plugin>  <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.5.2</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <createDependencyReducedPom>false</createDependencyReducedPom> </configuration> </execution> </executions> </plugin> </plugins> </build> </project> The Neo4j dependency is provided scope — it won't be bundled into the jar since Neo4j supplies it at runtime. The DuckDB JDBC driver is included at compile scope so it gets bundled along with its native binary. The JUnit dependency is test scope so it's excluded from the fat jar. The maven-shade-plugin builds the fat jar at package time. Confirm Maven can resolve the dependencies: Shell mvn dependency:resolve The Java Classes Create DuckDBConnection.java at src/main/java/com/example/duckdb/: Java package com.example.duckdb; import java.sql.Connection; import java.sql.DriverManager; import java.sql.SQLException; public class DuckDBConnection { private static volatile Connection instance; private DuckDBConnection() {} public static Connection getInstance() throws SQLException { if (instance == null || instance.isClosed()) { synchronized (DuckDBConnection.class) { if (instance == null || instance.isClosed()) { instance = DriverManager.getConnection("jdbc:duckdb:"); } } } return instance; } } The volatile keyword ensures the instance is visible across threads without full synchronization on every read. The jdbc:duckdb: connection string with no path gives us an in-memory DuckDB instance. Now create DuckDBProcedure.java in the same package: Java package com.example.duckdb; import org.neo4j.procedure.Description; import org.neo4j.procedure.Name; import org.neo4j.procedure.Procedure; import java.sql.Connection; import java.sql.ResultSet; import java.sql.ResultSetMetaData; import java.sql.Statement; import java.sql.Types; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.stream.Stream; public class DuckDBProcedure { public static class QueryResult { public Map<String, Object> row; public QueryResult(Map<String, Object> row) { this.row = row; } } @Procedure(name = "duckdb.query") @Description("Run a DuckDB SQL query and return results as rows") public Stream<QueryResult> query(@Name("sql") String sql) throws Exception { List<QueryResult> results = new ArrayList<>(); synchronized (DuckDBConnection.class) { Connection conn = DuckDBConnection.getInstance(); try (Statement stmt = conn.createStatement()) { boolean hasResultSet = stmt.execute(sql); if (hasResultSet) { try (ResultSet rs = stmt.getResultSet()) { ResultSetMetaData meta = rs.getMetaData(); int columnCount = meta.getColumnCount(); while (rs.next()) { Map<String, Object> row = new HashMap<>(); for (int i = 1; i <= columnCount; i++) { String colName = meta.getColumnName(i); int colType = meta.getColumnType(i); String typeName = meta.getColumnTypeName(i); Object value = mapValue(rs, i, colType, typeName); row.put(colName, value); } results.add(new QueryResult(row)); } } } else { Map<String, Object> status = new HashMap<>(); status.put("status", "OK"); status.put("updateCount", (long) stmt.getUpdateCount()); results.add(new QueryResult(status)); } } } return results.stream(); } private Object mapValue(ResultSet rs, int index, int sqlType, String typeName) throws Exception { if (typeName != null && typeName.equalsIgnoreCase("HUGEINT")) { long val = rs.getLong(index); return rs.wasNull() ? null : val; } switch (sqlType) { case Types.TINYINT: case Types.SMALLINT: case Types.INTEGER: { long val = (long) rs.getInt(index); return rs.wasNull() ? null : val; } case Types.BIGINT: { long val = rs.getLong(index); return rs.wasNull() ? null : val; } case Types.FLOAT: case Types.REAL: case Types.DOUBLE: case Types.NUMERIC: case Types.DECIMAL: { double val = rs.getDouble(index); return rs.wasNull() ? null : val; } case Types.BOOLEAN: case Types.BIT: { boolean val = rs.getBoolean(index); return rs.wasNull() ? null : val; } case Types.NULL: return null; default: return rs.getString(index); } } } A few things worth noting about DuckDBProcedure: We use stmt.execute(sql) rather than executeQuery() because DuckDB write operations like COPY don't return a ResultSet. The boolean return value tells us which path to take.The mapValue method handles type mapping explicitly. DuckDB's HUGEINT type – used for aggregated sums — has no standard java.sql.Types mapping, so we check the type name directly. Every numeric read is followed by rs.wasNull() to avoid returning 0 for a null column.We synchronize on DuckDBConnection.class to keep concurrent Cypher calls thread-safe against the shared connection. Writing the Tests We tried neo4j-harness first, which spins up a lightweight embedded Neo4j instance in JUnit — the same approach used in the Neo4j Procedure Template. In our environment, we found that the embedded server took longer than expected to become ready during the test run, which led us to a simpler alternative. The fix is to skip neo4j-harness entirely and test the DuckDB logic directly via JDBC. Since DuckDBProcedure is a thin wrapper around a JDBC connection, we can exercise the same type mapping, null handling, and file reading logic without Neo4j in the loop at all. The tests run in under 4 seconds and are fully self-contained. First, add the test directory structure and a src/test/resources folder to hold the test data files: Shell mkdir -p src/test/java/com/example/duckdb mkdir -p src/test/resources The test data files cover all the types and edge cases we want to verify: Plain Text src/test/resources/ ├── test.csv ├── test_nulls.csv ├── test_floats.csv ├── test_booleans.csv ├── test_timezones.csv └── test.json Maven copies everything in src/test/resources to target/test-classes automatically, so the test class can resolve file paths via the classloader without hardcoding any absolute paths. The test class at src/test/java/com/example/duckdb/DuckDBProcedureTest.java covers 12 cases: Range queriesCSV readingParquet write and readJSON readingAggregation returning numeric values rather than stringsNull handlingFloatsBooleansEmpty result setsBad SQL error surfacingIn-memory table persistence across callsDuckDB-side JOIN between two files Each test mirrors something we verified manually in Neo4j. File paths are resolved portably so the tests work regardless of where the project is cloned: Java private String resourcePath(String filename) { return getClass().getClassLoader().getResource(filename).getPath(); } The Parquet test is the one exception — since DuckDB generates the Parquet file rather than reading a pre-existing one, it writes to a JUnit @TempDir that's cleaned up automatically after the test run. Run the full suite with: Shell mvn test All 12 tests should pass in a few seconds. Running mvn package will also run the tests before building the fat jar, so any regression in the DuckDB logic is caught before the jar is deployed. Building and Installing Build the fat jar: Shell mvn package -DskipTests Any warning about MANIFEST.MF from the shade plugin is harmless. The jar lands at target/neo4j-duckdb-procedure-1.0.0.jar. Copy it to your Neo4j Desktop plugins folder. You can find the path in Neo4j Desktop under the three-dot menu for your database instance, then Open Folder > Plugins. Then copy the jar: Shell cp target/neo4j-duckdb-procedure-1.0.0.jar "/path/to/your/neo4j/plugins/" Open neo4j.conf in the conf folder alongside plugins and add: Plain Text dbms.security.procedures.unrestricted=duckdb.* Restart the database. Confirm the procedure is registered using Query from the left-hand navigation pane: Cypher SHOW PROCEDURES YIELD name WHERE name STARTS WITH 'duckdb' RETURN name You should see duckdb.query. If the procedure doesn't appear after restarting, also add the following in neo4j.conf: Plain Text dbms.security.procedures.allowlist=duckdb.* Some Neo4j configurations require procedures to be explicitly allowlisted before they're loaded at startup. Putting It to Work The test data files used here are also available in src/test/resources in the project repository. Let's start simple. DuckDB has a built-in range function we can use without any external data: Cypher CALL duckdb.query("SELECT range AS num FROM range(1, 6)") YIELD row RETURN row.num AS num Five rows should be returned, confirming that DuckDB is running inside Neo4j. Note that DuckDB's range() function is exclusive of the upper bound, so range(1, 6) returns 1 through 5. Now let's read a CSV file. Create a test file at /tmp/test.csv: Plain Text city,country,population London,UK,9648110 Paris,France,2161000 Berlin,Germany,3645000 Tokyo,Japan,13960000 New York,USA,8336817 Query it from Cypher: Cypher CALL duckdb.query("SELECT city, country, population FROM read_csv_auto('/tmp/test.csv') ORDER BY population DESC") YIELD row RETURN row.city AS city, row.country AS country, row.population AS population DuckDB reads and parses the CSV; Neo4j receives the result. We can also write the CSV out as a Parquet file: Cypher CALL duckdb.query("COPY (SELECT * FROM read_csv_auto('/tmp/test.csv')) TO '/tmp/test.parquet' (FORMAT PARQUET)") YIELD row RETURN row And read it back: Cypher CALL duckdb.query("SELECT city, country, population FROM read_parquet('/tmp/test.parquet') ORDER BY population DESC") YIELD row RETURN row.city AS city, row.country AS country, row.population AS population DuckDB handles JSON just as easily. Create /tmp/test.json: JSON [ {"city": "London", "country": "UK", "population": 9648110}, {"city": "Paris", "country": "France", "population": 2161000}, {"city": "Berlin", "country": "Germany", "population": 3645000}, {"city": "Tokyo", "country": "Japan", "population": 13960000}, {"city": "New York", "country": "USA", "population": 8336817} ] And read it: Cypher CALL duckdb.query("SELECT city, country, population FROM read_json_auto('/tmp/test.json') ORDER BY population DESC") YIELD row RETURN row.city AS city, row.country AS country, row.population AS population We can also join two files entirely inside DuckDB before the result reaches Cypher. Create /tmp/test_timezones.csv: Plain Text city,timezone London,GMT Paris,CET Berlin,CET Tokyo,JST New York,EST Then join it against the population data: Cypher CALL duckdb.query("SELECT p.city, p.population, t.timezone FROM read_csv_auto('/tmp/test.csv') p JOIN read_csv_auto('/tmp/test_timezones.csv') t ON p.city = t.city ORDER BY p.population DESC") YIELD row RETURN row.city AS city, row.population AS population, row.timezone AS timezone The join happens entirely inside DuckDB. Neo4j just receives the final result set. Now for the most interesting demo: combining DuckDB analytics with Neo4j graph data in a single query. First, create some graph nodes: Cypher UNWIND [ {city: "London", continent: "Europe"}, {city: "Paris", continent: "Europe"}, {city: "Berlin", continent: "Europe"}, {city: "Tokyo", continent: "Asia"}, {city: "New York", continent: "North America"} ] AS data MERGE (c:City {name: data.city}) SET c.continent = data.continent Now join DuckDB file data with Neo4j graph data by city name: Cypher CALL duckdb.query("SELECT city, population FROM read_parquet('/tmp/test.parquet')") YIELD row MATCH (c:City {name: row.city}) RETURN c.name AS city, c.continent AS continent, row.population AS population ORDER BY row.population DESC And aggregate by continent — population figures from DuckDB, continent groupings from the graph: Cypher CALL duckdb.query("SELECT city, population FROM read_parquet('/tmp/test.parquet')") YIELD row MATCH (c:City {name: row.city}) RETURN c.continent AS continent, count(c) AS city_count, sum(row.population) AS total_population ORDER BY total_population DESC That's two engines contributing to a single result: DuckDB handles the file analytics, Neo4j handles the graph traversal and aggregation. What We Learned Building this wasn't entirely smooth, and the bumps are worth documenting. XML comments in a Maven POM can't contain double dashes anywhere except at the closing -->. That's a quirk of XML parsing that's easy to miss.DuckDB's HUGEINT type – used for the result of aggregation functions like SUM – has no corresponding entry in the standard java.sql.Types constants. Our initial type mapping switch statement silently fell through to the string default, returning numeric values as strings. The fix is to check ResultSetMetaData.getColumnTypeName() by name before entering the switch.Write operations like COPY can't go through executeQuery() since they don't return a ResultSet. Switching to stmt.execute() and checking the boolean return value handles both read and write paths cleanly.Null numeric values in JDBC return 0 from getInt() and getLong(), not null. You must call rs.wasNull() immediately after each numeric read to distinguish a genuine zero from a missing value. Comparing Approaches It's also worth noting that APOC (Awesome Procedures on Cypher) Extended's apoc.load.jdbc procedure supports DuckDB via its JDBC driver and apoc.jdbc.analytics goes further by letting you export graph data into a temporary DuckDB table for analytical queries. These are complementary approaches worth exploring alongside the custom procedure built here. The following table compares the three ways to use DuckDB with Neo4j, each with a different integration style. ApproachHow It WorksDirectionPersistent StateRequiresCustom procedure (this article)DuckDB runs in-process inside the Neo4j JVM as a singleton connectionFiles and SQL into Neo4jYes, across Cypher callsCustom Java pluginAPOC apoc.load.jdbcDuckDB treated as an external JDBC data source; driver jar dropped into plugins folderFiles and SQL into Neo4jNoAPOC Extended + DuckDB JDBC jarAPOC apoc.jdbc.analyticsGraph data exported into a temporary DuckDB table; analytical SQL runs against itGraph data into DuckDB for analyticsNoAPOC Extended + DuckDB JDBC jar Summary We started with a simple question: could DuckDB be embedded in Neo4j the way pg_duckdb embeds it in Postgres? The answer is not natively, but closer than you might expect. With two Java classes, a fat jar, and a few lines of Neo4j configuration, we built a working duckdb.query procedure that runs DuckDB SQL directly from Cypher. Along the way we handled CSV, Parquet and JSON files, DuckDB-side joins, cross-engine aggregations with Neo4j graph data and a handful of type mapping edge cases that the 12-test suite now catches automatically. There's plenty of room to grow this further. File-backed DuckDB, a proper connection pool, cleaner error handling, and possibly a maintained open-source plugin are all natural next steps. But the foundation is there. What we've built here is a proof of concept — it works, and the tests confirm it, but anyone taking it further should treat robustness, connection pooling, and error handling as open items before putting it near production data. The full source code is available on GitHub.

By Akmal Chaudhri

CORE

Engineering Audit-Ready Data Pipelines for Regulatory Reporting

There's a particular kind of silence that happens on a call when a regulator asks, "Can you walk me through exactly how this number was calculated?" and the person on the other end has to say, "Let me get back to you." I've been on that call. It is not a fun place to be, and it usually isn't because anyone did anything wrong. It's because the pipeline that produced the number was never built to answer that question in the first place. Regulatory reporting, whether it's a liquidity return, a capital calculation, or something more niche, has a strange property compared to most data engineering work. Nobody outside the team really cares how the dashboard behind it looks. What they care about is whether you can prove, months or years later, exactly where every figure came from, who touched it, and why it changed between two reporting periods. That's a fundamentally different design goal than "make the report run fast," and honestly, a lot of pipelines get built for speed and convenience first, with auditability bolted on afterward as an afterthought. It shows. What "Audit-Ready" Actually Means in Practice I think people sometimes hear "audit-ready" and picture some heavyweight compliance layer sitting on top of a normal pipeline. In my experience, it's less about adding a layer and more about changing a handful of default behaviors early, before the pipeline exists in production. Four things tend to matter most. First, nothing gets overwritten. Ever. If a source record changes, you keep the old version and the new one, both timestamped. Second, every transformation step has to be reproducible from code, not from a person's memory of what they clicked in a tool three years ago. Third, there's a reconciliation point, a moment where the pipeline checks its own output against the source and either passes or stops. And fourth, a human actually looks at the number before it leaves the building. Not a dashboard glance, an actual documented sign-off with a name and a timestamp attached to it. Here's roughly how that fits together end to end. A couple of things in that picture are doing more work than they look like they are. The raw zone sitting right after ingestion, untouched and append-only, is what lets you go back eighteen months later and reconstruct exactly what the source system handed you on a given day, independent of whatever the curated tables look like now. I've genuinely had to do this, pulling raw records from a specific date to settle a dispute about whether a discrepancy came from us or from upstream. Without that raw layer, that conversation becomes a lot more uncomfortable. The Reconciliation Step Is Where Trust Actually Gets Built If I had to pick one part of this architecture that matters more than the rest, it's the reconciliation checkpoint. This is the moment where the pipeline stops behaving like a normal ETL job and starts behaving like a control. The logic itself isn't exotic. You compare row counts, sums, and control totals against the source, and you either match within tolerance or you don't. What matters is what happens on the "no" branch. A lot of pipelines I've inherited over the years treated a reconciliation break as a warning buried in a log file that nobody reads until someone complains. That's backward. A break should halt the pipeline and page someone, because a regulatory submission built on unreconciled data is worse than a late submission. I'd rather explain a delay than explain a restated figure. I'll admit there's a tension here worth naming honestly. Hard stops on every mismatch can grind a team down if your tolerance thresholds are too tight, and you end up with alert fatigue where people start ignoring pages because most of them turn out to be rounding noise from a currency conversion. Getting the thresholds right, informed by prior period baselines rather than guessed at, takes real tuning. It's not a solved problem you configure once and forget. Lineage and the Four-Eyes Check The governance band running underneath the pipeline, capturing lineage and version history, is what turns "I think this is right" into "here's the exact model, the exact commit, and the exact person who approved it." On Unity Catalog or an equivalent lineage tool, this is largely automatic once you've built your transformations in something like dbt rather than scattered SQL scripts. That's honestly one of the more underrated benefits of moving regulatory pipelines onto a tool with built-in lineage tracking. You stop having to build that tracking yourself. The human sign-off matters just as much, maybe more, even though it's the least technically interesting part of the whole design. A second person reviewing the output before submission, with their review logged, catches the kind of error that automated tests won't, like a number that's technically correct by every rule you wrote but obviously wrong to a person who understands the business context. A Small Piece of Advice If you're building one of these pipelines from scratch, resist the urge to treat the audit trail as something you'll add once the pipeline is "working." Retrofitting immutability and lineage onto tables that have already been overwritten a hundred times is painful, and in some cases you simply can't recover the history you needed. Build the raw, append-only layer first, even if it feels like overhead in week one. Six months later, when someone asks you to explain a number from last quarter, you'll be glad it's there.

By SUNILKUMAR REDDY Eraganeni

Moving 900+ GB of Prometheus Without a Cutover Weekend

The Prometheus I was responsible for had grown past 900 GB of time series data, and at that size it had every disease that comes with it. Queries timed out unless you narrowed them down to an hour. Cardinality kept growing, because teams added labels "for debugging" and then sometimes forgot to remove them. On top of that, we were shipping a copy of fractions of our metric data to an external SaaS vendor, just in case, and that bill grew every quarter. And the fleet under all of this ran latency-sensitive trading services, so monitoring itself was not allowed to eat an extra CPU core at a wrong moment. We moved it all to VictoriaMetrics. I am not going to argue here about storage engines; there is plenty written about which one wins. What nobody warned me about is how much the order of the steps matters. The standard way to migrate is a cutover weekend: you stand up the new system, you pick a Saturday, you switch, and then you spend three weeks finding out what broke. I think this order is backward. What worked for us was to delete first, then dual-write for a long time, then rebuild the dashboards and compare them panel by panel, and move the alerts at the very end. And most of the alerts should not move at all, but about that below. The Inventory Nobody Has Before touching anything, I wanted to know what anyone actually reads out of those 900+ GB. It sounds like a trivial question, and it is not, because the data you have does not answer it directly. The answer lives in three places. The TSDB status endpoint knows which metrics are the biggest. The Grafana dashboards know what humans look at. And the rule files know what the machines query, because alerts and recording rules read plenty of metrics which no human ever opens. The status endpoint reports on the in-memory head block, so strictly speaking it shows what is filling the disk going forward, and not the composition of the historical 900+ GB. As a proxy for the biggest offenders, it was good enough: Shell curl -s 'http://prometheus:9090/api/v1/status/tsdb?limit=15' \ | jq '.data.seriesCountByMetricName' For the second list, dashboards are JSON, so every expression can be pulled out with jq. I tokenized the expressions instead of parsing them, which means the output is polluted with PromQL function names and label names. The pollution disappears later, at the intersection step: Shell find dashboards/ -name '*.json' \ -exec jq -r '.. | objects | .expr? // empty' {} + \ | grep -oE '[a-zA-Z_:][a-zA-Z0-9_:]+' | sort -u > tokens_dashboards.txt The same jq pass over the exported alerting and recording rules gave the third list, and it caught a family of recording rules that fed an SLO report that no dashboard mentioned anywhere. (There is also mimirtool analyze, which builds this inventory for you. I wanted the raw lists.) Then you intersect the TSDB metric names with the union of the other two lists, and what falls out at the bottom is the set which nobody reads. For us that set covered a large share of the 900+ GB, and most of it was high-cardinality out of habit: per-pod histograms for services whose one dashboard had a single aggregate panel. Delete at the Source Before Anything Moves The instinct is to migrate everything and clean up later. Later is a bad deal here. Dead series which you keep get dual-written for months; they get baked into the sizing of the new cluster, and they pollute every panel diff which you will run later in the project. All of that in service of data which nobody reads. So the first change of the whole project landed on the old system, and not on the new one: YAML metric_relabel_configs: - source_labels: [__name__] regex: 'go_gc_duration_seconds.*|.*_bucket_experimental_.*' action: drop - source_labels: [__name__, pod] regex: 'legacy_worker_queue_depth;.+' action: drop The obvious worry is that you will need a deleted metric later. That came up twice over the following weeks, and both times the fix was to remove one relabel rule and reload the config with a SIGHUP or a POST to /-/reload. Do not restart a 900 GB Prometheus for this; the WAL replay will leave you without monitoring for longer than you expect. And one caveat which you should agree on before deleting anything: the reversal only works going forward. The rule drops samples before ingestion, so the gap for the dropped period never backfills. In our case nobody ever missed those gaps. The Dual-Write Months With the estate slimmed down, the old Prometheus started writing into the new storage in parallel with its own: YAML remote_write: - url: http://vminsert.internal:8480/insert/0/prometheus/api/v1/write queue_config: max_samples_per_send: 10000 capacity: 50000 max_shards: 12 Those numbers were not the first ones we tried. With the defaults, the send queue was too shallow for our sample rate, and within the first hour the log was full of this: Plain Text ts=... caller=queue_manager.go level=warn component=remote msg="Remote storage falling behind" pending_samples=48231 shards=4 So max_samples_per_send went up, capacity followed at five times that, which is what the tuning guidance says, and shards were raised until the queue stopped falling behind at the evening peak. And because the trading fleet was the whole point of being careful, we measured what dual-writing costs the old server before trusting it. CPU rose by roughly a tenth. Memory rose by about what the queue capacity implies. Scrape latencies did not move. The abort criterion was written down in advance: if scrape or rule evaluation starts lagging, the remote_write block comes out that same day. It never triggered. Then, deliberately, nothing else happened for roughly a quarter. The old system stayed the source of truth for every dashboard and every page, and the new one filled up in the background, until every window which people habitually query existed on both sides, from the last six hours out to the month-over-month capacity view. This is the whole point of dual-writing: at no moment did the project depend on one switch being flipped correctly on one Saturday. The one incident which the phase produced was self-inflicted. A relabel change on the old side renamed a label on its way into the new storage only, and the affected series split in two. A panel diff caught it about a week later, and that is exactly the place where such things should get caught. Dashboards Are the Test Suite We rebuilt the dashboards against the new engine while the old ones kept running, and then compared them panel by panel over the same time ranges. Every discrepancy meant something, but not always that the new panel was wrong. The classic one is increase(). Prometheus extrapolates it, so a counter which grew by exactly 300 over the window renders as 299.4 or 301.7, and everyone has learned to not see it anymore. MetricsQL does not extrapolate, so the same panel suddenly shows round integers: Plain Text increase(http_requests_total{service="order-gateway"}[1h]) # old panel: 299.4 new panel: 300 Some panels had been quietly misleading for years, and the diffs surfaced that. Other panels were wrong on my side: a rate window that did not match the scrape interval, or a sum that collapsed a label that the legend needed. The comparison caught both kinds. The work was tedious. It was also the cheapest reliability work of the whole project, because whatever you catch at the desk at this stage, nobody has to chase later in the middle of an incident. Alerts Move Last, and Most Should Not Move Alerting stayed on the old system until everything above was finished. A bad page wakes somebody up at night, so pages moved in the last phase, group by group, and every rule was re-run against the new engine and watched for a while before its old copy retired. The stronger claim, and the one I get pushback on: a migration is the wrong moment to preserve your alert rules, because most of them do not deserve the trip. Ours had accumulated over years; they fired constantly, and they meant little. Copying them onto the new engine would carry the same noise across intact, just on faster storage. So the rules were rebuilt around a single test instead: does this page fire on something which a user or a dependent service can feel. YAML # Retired with the old system: - alert: HighCPU expr: 100 - (avg by(instance) (rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100) > 90 for: 10m # What replaced it: - alert: OrderGatewayP99LatencyHigh expr: histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket{service="order-gateway"}[5m])) by (le) ) > 0.4 for: 5m CPU sitting at 91 percent has never made anyone I know roll a deploy back. The rollbacks which I have actually watched were all triggered by latency which a client could feel, and that is what the replacement rule fires on. The rest — the CPU levels, the queue depths, the disk growth — went into dashboards and recording rules for capacity planning. And outside of a migration, deleting a single alert rule means an argument with whoever added it. The migration made bulk deletion politically possible for once, and roughly a hundred stale rules did not make the trip. What Fell Out of It After the source-side deletes, the estate which actually needed a new home was well under half of the original 900+ GB. The SaaS dependence went away, and the bill went away with it. Queries stopped timing out. And the pages dropped from dozens on a loud week to a handful, so the ones which did fire got taken seriously, and incident response got faster off the back of that. The migration ran for months under a live trading workload and produced no cutover incidents, because at every step the previous state was still there, and we could fall back to it. If you are staring at your own 900 GB, the short version is this. Run the inventory and believe it, even when it embarrasses the dashboards which produced it. Delete at the source before you move a byte. Dual-write across the longest window that people actually query. And when the time comes to move the alerts, ask about every rule whether a user can feel the thing that it fires on, because a migration is the only moment when your organization will agree to delete alert rules in bulk. A cutover weekend takes two days, and this took months; that is true. But the weekend does not remove any of the work that is described above. It just moves the increase() mismatches and the stale rules into production, and there they surface one at a time, usually at night.

By Daniil Romashov

Data

DZone's Featured Data Resources

Monthly Top Data Experts

The Latest Data Topics