Big Data

Dashboards are everywhere. Business and IT teams use them to track metrics, visualize trends, and make decisions. But when working with real-time data from Apache Kafka, it’s not obvious how to connect dashboards to the stream or whether you should at all. The conversation often jumps to technical options like Flink SQL, Kafka Streams Interactive Queries, or Confluent's TableFlow. Others try to build interactive dashboards directly on top of Kafka topics using a JDBC connector into a database and a Business Intelligence tool. But that only makes sense once the actual goal is clear. What is the business trying to do with the data? Dashboards are not always the right tool. Automation, smart agents, or process intelligence often deliver more value. Let’s unpack the bigger picture. This blog post breaks down the different types of queries on Apache Kafka data, when dashboards make sense, and why a context engine often plays a key role. Why Dashboards — And When Not To Use Them Dashboards give people visual access to data. They support decisions, reporting, and oversight. But not all data needs to be visualized. Dashboards make sense when: Business users want a regular view of changing dataTeams need to investigate operational metricsThere is a requirement for manual filtering and inspection But in many cases, dashboards are not the best answer. For example A machine overheating should trigger an alert, not wait for someone to look at a graphA fraud detection system should act instantly, not visualize the anomalyAn AI agent monitoring supply chains should get structured context, not a dashboard snapshot In these scenarios, dashboards are a fallback. The real need is action or automation, not visualization. This is where agentic AI and process intelligence come into play. AI agents require structured, fresh context. They do not use dashboards. They consume streaming data, apply logic or reasoning, and trigger downstream actions. Dashboards might still be used to audit what happened but not to drive the process itself. So before jumping into dashboard tools, first ask: Is this data for a human to observe or a system to act on? Foundations First: Apache Kafka, Event Streaming, Data Products, and Governance Apache Kafka is the core of modern event-driven architecture. It enables systems to stream events in real time, such as customer interactions, machine signals, backend transactions, or system logs. Unlike batch pipelines, event streaming allows continuous data flow across the business. This supports responsive applications, automation, and real-time analytics. But fast data is not enough. Real-time value depends on reliable data. That’s why many teams now treat Kafka topics as data products. Each stream should have a clear owner, a defined schema, and a contract between producers and consumers. Schemas must be versioned and validated. Metadata must be consistent and available. Lineage, access control, and quality checks are critical to avoid downstream errors. Without this foundation, queries will return incorrect results, and automation may act on bad signals. Governance, schema control, and product thinking are not extras. They are required to build trustworthy systems on streaming data. Three Kinds of Queries for Apache Kafka Events If a dashboard is needed, the next step is understanding the type of query behind it. This helps define the right technical setup. Operational Queries These are fully automated. They respond to events and trigger actions. Think of them as the nervous system of an application. They are built directly into stream processing applications using Apache Flink or Kafka Streams. The logic is reactive and runs continuously. These systems are part of mission-critical operations. They must be highly available, fault-tolerant, and operate with minimal latency. Any downtime or delay can disrupt core business processes. A modern data streaming platform that augments Kafka and Flink with on-the-fly table serving, snapshot queries, and a context engine helps close this gap between streaming and interactive exploration. Example use cases include raising alerts on thresholds, aggregating orders for reporting, or triggering workflows. These systems should not rely on dashboards. Explorative Queries These are used by people to explore the data. They are ad hoc, flexible, and interactive. This type of query is difficult to support directly on Apache Kafka. Kafka is optimized for high-throughput event streaming and acts as an immutable event log. It provides a durable persistence layer and decouples producers from consumers, which makes it ideal for data pipelines and ensuring data consistency across real-time and batch systems. However, it is not designed for indexed lookups or ad hoc filtering across large datasets. Kafka does not offer queryable storage, secondary indexes, or snapshot consistency, all of which are essential for interactive exploration. Flink can process the data, but it does not offer indexed access. That makes joins or drilldowns inefficient without an external engine. Exploratory queries are often run in SQL workbenches, BI tools like Superset, or analytical engines like Druid and ClickHouse. They are useful for finding anomalies, trying out new logic, or investigating correlations. They require indexing, snapshot consistency, and historical access. Example use cases include joining marketing and sales events to find conversion patterns, analyzing user journeys through digital platforms, or testing new business rules across historical data. These queries typically require interactive tools and should not rely on stream processing systems alone. Monitoring Dashboards This use case is simpler but more common. The goal is to display filtered, consistent, and up-to-date data to end users. It does not involve complex joins or deep exploration. Instead, dashboards show metrics from recent data, business KPIs, or precomputed aggregations. Tools used here include Power BI, Grafana, or custom frontends connected to Flink or TableFlow. Dashboards in this case should be thin and rely on upstream systems for logic. Example use cases include showing live production status on a factory screen, displaying transaction volumes in a finance dashboard, or visualizing the health of streaming pipelines for operations teams. These dashboards are read-only and should not contain business logic. What Businesses Really Need Today While use cases vary, a few patterns repeat across industries. These needs can guide architecture decisions. Lightweight dashboards with filtering but no complex joins: Power BI and Grafana are the most common tools. Used for message tracing, monitoring, and status overviews. Users prefer querying externally instead of importing data.Real-time data that stays up to date: Dashboards refresh automatically. Data is pushed from Flink or precomputed topics. Materialized views support this, but changing schema can cause frontend problems.Business logic belongs upstream: Dashboards should not do computation. Flink or Kafka Streams handle the logic and prepare the data.Integration with ML models and agents: Dashboards may show results from predictions or scoring models. These are often trained ML models, not LLMs. Model drift monitoring is gaining interest. LLMs are still early stage in these setups.Protocol-agnostic connectors: REST, WebSocket, MQTT, JDBC — all needed. Most organizations expect flexible integration. Sink connectors alone are often not enough. APIs with query parameters are common requests. The Context Engine: Serving Dashboards and AI Agents from Apache Kafka Events A powerful pattern is the context engine. It connects Kafka streams to dashboards and AI systems by offering real-time, structured, and indexed access to data. It works like this Flink or Kafka Streams process raw Kafka topicsOutput data flows to context topicsA service builds indexed views of relevant business objectsDashboards and agents query those views through an API This setup creates a reliable source of truth. Business logic stays in the stream. The context engine focuses on enrichment, access control, and exposing views. For AI agents, this API layer usually follows the Model Context Protocol (MCP), which is becoming the de facto interface for connecting agents to structured enterprise data. Dashboards, in contrast, are typically served from materialized views in cache or in-memory databases, or directly through REST APIs optimized for low-latency reads. Agentic AI systems benefit directly. They consume these views as context to make decisions in real time. Instead of querying raw data or relying on stale batches, they get structured signals. Generative AI also benefits, using the same views as grounding data. Dashboards and AI agents both rely on fresh, accurate context. A context engine provides that bridge. Start With the Use Case, Not the Tool The right dashboard architecture does not start with a tool choice. It starts with business needs. Ask the right questions: What decisions or actions should this data support?Is the goal observation or automation?Does the user need filtering, drilldowns, or live KPIs?How fresh must the data be?Can the logic run upstream, or must it remain flexible? These answers will guide the setup. Sometimes a simple Power BI dashboard is enough. Other times a context engine or Flink job is required. In many cases, a dashboard is just the user interface to something much more powerful running behind the scenes. Of course, even when the focus is on business outcomes, a tool still has to be selected. That decision should follow the use case, not drive it. There are many options. Some teams prefer code-driven frameworks that give full control and allow deep integration with APIs and AI agent interfaces. Others choose no-code or low-code tools with prebuilt widgets so business users can create interactive views quickly. Each option comes with trade-offs in flexibility, governance, scalability, and integration. Exploring these tooling choices in depth would fill an entire chapter on its own. The key message here is simple: start with the outcome. The tool is an implementation detail. Build for the decision, not for the visualization. That is how streaming data creates real business value.

By Kai Wähner CORE

For the past two decades, most enterprise data engineering systems have been built on one default assumption: People understand the system. The system executes the pipeline. Engineers understand the business context, break a requirement into steps, write SQL, Spark jobs, shell scripts, synchronization tasks, and scheduling workflows, and then let the system run them. The scheduler does not need to understand the business. The sync engine does not need to understand the metric. It only needs to execute the predefined flow reliably. That model supported the era of data warehouses, data lakes, BI reporting, and batch scheduling very well. But now that assumption is starting to break down. Enterprise data systems are becoming more complex in every direction: More data sources.Longer pipelines.Stronger real-time requirements.Faster business changes.More conflicting metric definitions.More AI application data, model feedback data, vector indexes, and unstructured content. In this environment, enterprises do not just need more pipelines, and they do not just need a better Copilot that can write SQL faster. They increasingly need a Data Engineering Agent that can understand the system, plan tasks, call tools, validate outcomes, and accumulate experience over time. In that shift, Apache SeaTunnel becomes especially important. Because in the agent era, it is not enough for a system to "think." It also has to connect to real data sources, capture changes, execute synchronization, process incremental updates, preserve consistency, and move data to target systems in a reliable and cost-effective way. In other words: The agent understands the goal and plans the action. SeaTunnel turns that action into real, reliable, and recoverable data movement. That is why SeaTunnel is well positioned to become a core execution foundation in the evolution from ETL, ELT, and EtLT to agent-driven data engineering. ETL to ELT: The First Major Shift Traditional ETL is straightforward: Extract data from the source.Transform it in an intermediate layer.Load the processed result into the target system. This model fit the early data warehouse era well. At that time, data sources were relatively limited, the pipeline was easier to understand, and compute resources were more centralized. Enterprises wanted to clean the data, standardize the structure, and define the core logic before loading data into the warehouse. At its core, ETL is a deterministic pipeline model. Its key assumption is: People define the process in advance. The system executes the process. Later, with the rise of cloud warehouses, data lakes, lakehouse architectures, and elastic compute, ELT became more popular. ELT changed the order: ExtractLoadTransform inside the target platform. Instead of transforming everything before loading, enterprises started moving raw or near-raw data into a unified storage layer first, then using the target platform's compute power for downstream modeling and analytics. ELT solved several ETL limitations: It reduced upfront processing complexity.It preserved more original data.It gave analysts and modeling teams more flexibility later. But ELT also created a new problem. If all transformation is delayed until after loading, then dirty source data, schema drift, type mismatches, CDC events, privacy fields, and format inconsistencies all arrive directly in the target system. That might be acceptable in simple batch scenarios. It becomes much more expensive in real-time synchronization, CDC, multi-table sync, lakehouse ingestion, SaaS API ingestion, and AI-oriented data engineering. That is where a third pattern becomes more useful: EtLT. Why EtLT Matters EtLT is not just a compromise between ETL and ELT. A more useful way to understand it is: Extract -> lightweight transform -> Load -> semantic Transform That means: Extract the data.Apply the minimum engineering transformations required to make the data usable.Load it into a unified data foundation.Apply business-level and semantic transformation later. The key idea is the distinction between lowercase t and uppercase T. Lowercase t is not heavy business modeling. It is the engineering work that must happen before data enters the platform safely and consistently, such as: Field projectionType mappingFormat normalizationPrimary key or partition field handlingSensitive field maskingCDC event conversionMulti-table routingSchema evolution handlingPre-ingestion quality validationOne-read, multi-write patternsRate limiting and parallelism control. These transformations should not always be postponed to the target system. Otherwise, the lakehouse or warehouse becomes full of inconsistent, weakly governed, and semantically unclear raw data. At the same time, lowercase t should not try to absorb all business logic. Complex business definitions, KPI semantics, subject-area modeling, and cross-domain aggregation still belong to uppercase T, which should happen in the warehouse, lakehouse, semantic layer, or metric layer. That is the value of EtLT: Standardize the data engineering layer before loading, then apply business semantics after loading. This is exactly the place where SeaTunnel fits naturally. Its Source, Transform, and Sink architecture is well suited for the lowercase t in EtLT. It can connect heterogeneous systems, apply lightweight transformation during movement, handle CDC, adapt schemas, route multiple tables, and write the result into the target platform. In an EtLT architecture, SeaTunnel is not just a data mover. It becomes the data integration runtime that prepares data before it enters the unified data foundation. Why Traditional ETL Starts to Struggle Traditional ETL is built for relatively stable pipelines. You write the rules, draw the DAG, schedule the tasks, and fix failures when they happen. But modern enterprise data environments are no longer that simple. Today a single enterprise may operate across: OLTP databasesKafka streamsCDC pipelinesSaaS APIsObject storageLogs and eventsLakehouse platformsReal-time OLAP systemsVector databasesAI interaction logsModel output datasets. The problem is not only that there is more data. The data is also more fragmented, more heterogeneous, and more real-time. Pipeline length is another issue. A single business metric may depend on dozens of tables, multiple layers of wide tables, several business domains, and a long chain of definition changes. At that point, many enterprises no longer struggle with "Can we build the workflow?" They struggle with "Can anyone still explain the whole pipeline end to end?" This is where traditional ETL shows a structural limitation. One renamed field can break hundreds of tasks.One changed enum can silently shift multiple core metrics.One incorrect incremental logic branch can pollute an entire downstream analysis chain. The scheduler can tell you that a task failed. It usually cannot tell you why that task matters. The sync tool can move the data. It usually cannot tell you which business metric is now at risk. The engineer can fix the script. But only if that engineer can first rebuild the missing context. So the real weakness of traditional ETL is not just performance or reliability. It is that: It can execute the process, but it does not understand the system. Why Copilot is Not Enough Many teams first bring AI into data engineering through Copilot-style workflows: Generate SQLComplete Spark codeDraft YAMLProduce test samples. These capabilities are useful. They improve local productivity. But they do not solve the deepest problem in enterprise data engineering. Because the hardest part of data engineering is rarely just code generation. It is system understanding. Copilot can help generate a SQL statement, but it does not know the real business meaning of the field. It can help draft a synchronization task, but it does not know which downstream metrics will be affected by a schema change. It can help generate a scheduler config, but it does not know whether the change breaks historical consistency or recovery semantics. What enterprises actually struggle with includes: Lineage reasoningDependency analysisSemantic understandingMetric governanceRisk estimationImpact analysisIncremental recovery. These are not just autocomplete problems. So enterprises do not only need an AI IDE. They increasingly need an agentic data engineering system that can understand the target, decompose tasks, call engineering tools, and verify the result. The Real Shift: From Pipeline to Agent If we keep only one conclusion, it is this: Traditional ETL is "people define the process, systems execute the process." Agentic data engineering is "people define the goal, systems generate the process." That is not a slogan. It is a change in how work is organized. In the traditional model, engineers design the task chain first, configure Source, Transform, and Sink, and then let the scheduler execute the pipeline. The system faces a fixed process. In the agent model, the input may only be a business goal. For example: Add a new gross margin metric for orders and keep it aligned with the finance definition. Traditionally, the engineer must: Identify relevant data sources.Read table schemas.Inspect lineage.Design transformation logic.Configure sync and scheduling jobs.Add quality checks.Run regression validation. In an agent-oriented workflow, the system should be able to generate a sequence of actions around the goal: Identify the affected business entities.Discover candidate data sources.Analyze upstream lineage.Decide whether the job belongs to ETL, ELT, or EtLT.Generate or update the SeaTunnel synchronization task.Configure full-load or CDC mode.Apply lightweight transformation.Write the result into the warehouse or lakehouse.Trigger data quality validation.Evaluate downstream impact.Present the result for human confirmation. That is the real difference. The breakthrough is not "AI wrote a SQL statement for me." The breakthrough is: The system starts generating engineering actions from a business goal. But this immediately raises a critical question: When the agent plans a data action, who executes it reliably? That is exactly where SeaTunnel becomes essential. Apache SeaTunnel in the Agent Era: The Data Integration Execution Layer An agent cannot stop at reasoning and recommendations. If a Data Engineering Agent decides that a table should be synchronized, a CDC job should be adjusted, a broken pipeline segment should be replayed, or a data slice should be reloaded into the target system, it needs a stable and observable execution layer to carry out that decision. That execution layer needs several core capabilities. 1. It Must Connect to Many Kinds of Data Sources Enterprise data systems are inherently heterogeneous. An agent cannot live in a world with only one database or one file system. It needs to connect to MySQL, Oracle, PostgreSQL, SQL Server, Kafka, Hive, Iceberg, Doris, ClickHouse, StarRocks, Elasticsearch, S3, HDFS, MongoDB, and many other systems. SeaTunnel's connector architecture is designed for exactly this kind of environment. It abstracts Source, Transform, and Sink through a consistent plugin model so heterogeneous systems can be integrated in a unified way. 2. It Must Support Batch, Streaming, CDC, and Large-Scale Synchronization The agent era does not run on a single data movement pattern. It needs: One-time full migrationContinuous CDCOffline batch movementReal-time synchronizationSingle-table syncMulti-table or database-level sync. SeaTunnel is valuable here because it is not just a script wrapper for ETL. It is a real data integration runtime that can support full load, incremental sync, real-time processing, CDC, and multi-table movement in the same ecosystem. 3. It Must Handle the Lowercase t in EtLT Agentic systems do not need every business transformation to happen inside the sync layer. But they do need the sync layer to complete the minimum engineering transformation required to make the data trustworthy and usable before it lands in the platform. SeaTunnel's Transform layer is a strong fit for: Field mappingType conversionFilteringColumn projectionData maskingRoutingSimple reshaping. That is exactly the role of the lowercase t in EtLT: Do not overload the movement layer with heavy business modeling, but make the data governable and ready for the next stage. 4. It Must Provide Consistency, Fault Tolerance, and Recovery An agent can decide that a broken link should be replayed. But replay only matters if the underlying system can recover correctly. The execution layer still needs checkpointing, failure recovery, state handling, restart behavior, and strong delivery guarantees where needed. A reasoning layer without a reliable execution layer becomes a planner without hands. That is why execution quality still matters as much as intelligence. What the Future Stack Starts to Look Like If we look one step ahead, enterprise data engineering increasingly resembles a layered operating system rather than a collection of disconnected pipelines. In that stack: The semantic layer defines the business model.Metadata provides structure and context.Memory accumulates operational experience.The planning layer turns goals into actions.The execution layer performs synchronization, CDC, movement, replay, and recovery. SeaTunnel belongs to this execution layer. That placement is important. The future is not "put a large language model on top of ETL." The future is a coordinated system where reasoning and execution are separated clearly: The agent decides what should happen.SeaTunnel ensures that it actually happens in a reliable way. The Evolution in One Sentence ETL built data pipelines. ELT moved raw data into a unified platform first. EtLT rebalanced pre-load engineering standardization and post-load semantic modeling. The agent era pushes data engineering one step further: From fixed pipelines to goal-driven systems. In that world, SeaTunnel is not just a synchronization tool. It becomes a practical execution foundation for agentic data engineering. Agents make data systems understand goals. EtLT makes ingestion more controllable. SeaTunnel turns those goals into reliable data engineering actions. That is the deeper change now happening across enterprise data engineering.

A large API response becomes a client problem long before it becomes a network problem. A browser can receive hundreds of megabytes and still become unresponsive while buffering bytes, parsing one enormous JSON document, retaining duplicate object graphs, and rendering too much state on the main thread. The reliable solution is not a larger timeout. It is to stop treating the response as a synchronous document and start treating it as a durable, observable job whose data arrives in bounded pieces. Browser streams support incremental consumption and backpressure, while background workers allow long-running processing to remain independent of user-interface scripts. The Response Becomes a Job, Not a Payload The public API should acknowledge work quickly and return a stable job identifier rather than hold an HTTP connection open until every upstream page has been fetched. A 202 Accepted response establishes that contract without implying completion. The client can then subscribe to progress events, request a partial view, or retrieve a final artifact when the job reaches a terminal state. RFC 9110 defines 202 Accepted specifically for requests accepted for processing when processing has not necessarily completed. Java @PostMapping("/reports") public ResponseEntity<JobAccepted> create(@RequestBody ReportRequest request) { String jobId = UUID.randomUUID().toString(); workflowClient.start(reportWorkflow::run, jobId, request); return ResponseEntity.accepted() .header("Location", "/reports/" + jobId) .body(new JobAccepted(jobId, "QUEUED")); } This endpoint performs no large download or expensive transformation. It creates an addressable unit of work and returns immediately. The browser remains responsive because the initial response is tiny, while server capacity is protected from long-lived request threads. The job record should expose states such as queued, fetching, indexing, ready, failed, and canceled, with progress kept monotonic and coarse enough to remain trustworthy. Temporal Owns the Long-Running Control Flow Temporal fits the control plane because Workflow state survives process crashes and worker restarts, while failure-prone operations such as remote API calls belong in Activities with explicit timeouts and retry policies. Temporal documentation distinguishes deterministic Workflow logic from non-deterministic Activities and provides retry, timeout, heartbeat, and message-passing mechanisms for long-running execution. Java @WorkflowMethod public ResultRef run(String jobId, ReportRequest request) { String cursor = null; int sequence = 0; do { PageRef page = activities.fetchAndStore(jobId, cursor, sequence); activities.publishChunkReady(jobId, page); cursor = page.nextCursor(); sequence++; } while (cursor != null && !canceled); activities.buildIndex(jobId); activities.publishCompleted(jobId, sequence); return new ResultRef(jobId, sequence); } @SignalMethod public void cancel() { canceled = true; } Only references and counters should cross Workflow boundaries. Passing raw pages through Temporal causes every Activity input and result to accumulate in Event History. Temporal warns that large histories increase Workflow Task latency, documents a 50 MB or 51,200-event history limit, and recommends external storage plus Continue-As-New for large or long-running executions. The response body therefore belongs in object storage, while Temporal retains keys, checksums, cursors, and status. The fetching Activity should checkpoint often enough to support retries without restarting the transfer. Heartbeat details can carry the last committed cursor or byte range. Temporal recommends heartbeats for long-running Activities because missed heartbeats can trigger failure detection and retry. Java public PageRef fetchAndStore(String jobId, String cursor, int sequence) { UpstreamPage page = upstream.fetch(cursor); String key = storage.put(jobId + "/" + sequence, page.bytes()); Activity.getExecutionContext().heartbeat( new FetchCheckpoint(sequence, page.nextCursor()) ); return new PageRef( key, sequence, page.nextCursor(), page.sha256() ); } Kafka Carries Facts, Not Giant Documents Kafka is most effective as the event backbone, not as a substitute for object storage. Events should describe what happened and point to durable data, ChunkStored, ChunkIndexed, JobProgressed, JobCompleted, or JobFailed. Kafka enforces record-size limits at both producer and broker levels, so pushing multi-megabyte fragments into records creates brittle configuration coupling and expensive retries. Every event should use jobId as the key. Kafka partitions are ordered logs, and records sharing a key normally land in the same partition, preserving per-job sequence while allowing unrelated jobs to scale across partitions. Consumer groups distribute partitions across workers and rebalance them when membership changes. Java public void publishChunkReady(String jobId, PageRef page) { ChunkReady event = new ChunkReady( jobId, page.sequence(), page.storageKey(), page.sha256() ); kafkaTemplate.send("report-events", jobId, event); } Duplicate delivery must be assumed at every boundary. Kafka producer idempotence prevents duplicate writes caused by producer retries when compatible acknowledgment and in-flight settings are used, but downstream side effects still require idempotent consumers. An indexer can enforce uniqueness with (jobId, sequence, checksum) and commit its database transaction before acknowledging the Kafka offset. Backpressure should be expressed through bounded concurrency rather than hidden in memory. An Activity can publish one stored chunk at a time, while indexer lag indicates downstream pressure. Temporal can pause between pages when lag crosses a threshold, or consumers can scale until partition count becomes the limit. The Client Receives Progress and Bounded Content Server-sent events are sufficient when communication is primarily server-to-client. The protocol uses text/event-stream, keeps a persistent HTTP connection, and represents each notification as a small text block. A projection service can consume Kafka events, maintain the latest job state, and expose a resumable stream using application event IDs Java @GetMapping( value = "/reports/{jobId}/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE ) public Flux<ServerSentEvent<JobEvent>> events( @PathVariable String jobId) { return eventProjection.stream(jobId) .map(event -> ServerSentEvent.<JobEvent>builder() .id(event.sequence().toString()) .event(event.type()) .data(event) .build()); } The client should render status changes and small previews, not append the full raw response into application state. When direct streaming is required, the Fetch API exposes the response body as a ReadableStream, allowing chunk-by-chunk processing rather than waiting for completion. Parsing should occur incrementally, with CPU-heavy decoding or transformation moved to a Web Worker, whose execution remains separate from user-interface scripts. Final delivery should usually be a paginated query API, a range-readable artifact, or a signed download URL. A giant JSON reconstruction endpoint merely recreates the original failure at the last step. RAG Turns Stored Volume Into a Useful Interface RAG becomes valuable after chunks are durably stored. Each chunk can be normalized, split along semantic boundaries, embedded, and indexed with metadata containing the job identifier, source sequence, object key, and byte range. The original RAG formulation combines parametric generation with retrieved non-parametric memory, grounding generation in selected passages rather than the entire corpus. Java @KafkaListener( topics = "report-events", groupId = "rag-indexers" ) public void onChunkReady(ChunkReady event) { if (index.exists( event.jobId(), event.sequence(), event.checksum())) { return; } byte[] payload = storage.get(event.storageKey()); chunker.split(payload).forEach(chunk -> index.upsert( event.jobId(), event.sequence(), chunk ) ); progress.markIndexed( event.jobId(), event.sequence() ); } The query path retrieves only the most relevant chunks and sends those bounded passages to the model. Raw object references remain attached so generated statements can link back to source material. RAG should not conceal incomplete ingestion; the query service must expose index coverage and reject complete-report requests until all expected chunks are indexed. Java public Answer answer(String jobId, String question) { List<Passage> context = index.search(jobId, question, 8); return generator.generate(question, context); } This layer changes the client experience from downloading everything before anything is useful to inspecting progress, searching partial results, and retrieving only relevant evidence. It also keeps model context bounded when the source response is extremely large. A Responsive System Is Built From Explicit Boundaries The essential boundary is simple: Temporal owns durable intent and recovery, Kafka distributes compact facts, object storage holds large bytes, RAG builds a searchable semantic view, and the client receives only bounded updates or explicitly requested slices. Each component solves a different failure mode, and none is forced to carry the complete response through an interface designed for small messages. The resulting architecture prevents UI freezes, survives retries and restarts, supports cancellation and replay, and makes large upstream results useful before a monolithic download could finish. Large-response handling becomes reliable when completion is modeled as a process rather than a payload.

By Uthej Mopathi CORE

I learned this lesson the hard way. We had a critical data pipeline running for over 3 hours every single day. The logic was perfectly clean. The overarching schema was explicitly right. There were absolutely no obvious memory leaks, and absolutely nothing looked fundamentally broken in the raw PySpark transformations. Then I finally checked the physical query plan. Under the hood, Apache Spark was quietly executing a massive Sort-Merge Join to merge a multi-terabyte fact table with a dimensional lookup table that was barely 50MB in total size. One single line of code changed—wrapping that exact tiny lookup table cleanly in a broadcast() hint—and the exact same analytic job plummeted from 3 hours to just 18 minutes. That was it. One word saved us hours of daily compute and massive underlying cloud FinOps costs. Wrong join types are financially devastating directly because they are completely silent. Spark will not throw an aggressive exception. Your pipeline will not explicitly fail. It will simply execute your logic confidently 10× slower than it architecturally ever needs to. Here is the exact mental model I exclusively use now every single time I write a distributed join in Apache Spark. TL;DR: Silent shuffle bottlenecks kill massive Spark performance. Always explicitly broadcast small tables (< 200MB), default natively to Sort-Merge for massive dual-sided joins, and actively, aggressively leverage AQE skew joins to fundamentally prevent heavy task skew. Check your physical plans! 1. The Small Table: Always Broadcast When actively joining a massive fact table heavily against a tiny dimension table (like cleanly mapping a primitive status_id logically to a status_name), globally shuffling the multi-terabyte fact table wildly across the distributed cluster is architectural suicide. The rule: If a table is reliably under 200MB, rigorously physically force a Broadcast Hash Join.The mechanism: Spark intelligently bypasses the massive network shuffle entirely. It simply naturally copies the tiny 50MB table directly into the RAM of every single native worker node, allowing them to map data logically and locally. The Implementation Python from pyspark.sql.functions import broadcast # Wrapping the small lookup table strictly natively in a broadcast hint enriched_df = massive_fact_df.join( broadcast(small_lookup_df), "customer_id", "left" ) 2. Both Sides Massive: Default to Sort-Merge If you are systematically actively joining two massive, multi-terabyte tables accurately together (e.g., dynamically merging historical transactions cleanly with historical web_sessions), you physically cannot organically broadcast data without instantly dynamically triggering brutal Out-Of-Memory (OOM) driver exceptions. The Rule: Default heavily unconditionally to the Sort-Merge Join.The Mechanism: This is Spark's absolute most robust, incredibly stable joining algorithm physically built for massive scale. Spark heavily and organically shuffles the massive data wildly across the cluster so that precisely matching keys uniquely land physically on the exact same nodes, fundamentally and strictly sort them, and actively, efficiently, and accurately merge them natively. It is technically slower than a pure broadcast, but it is incredibly beautifully resilient inherently at petabyte scale. The Implementation Python # No explicit hints structurally required. Spark will cleanly natively default seamlessly to Sort-Merge for massive large datasets. final_df = massive_transactions_df.join( massive_sessions_df, "user_id", "inner" ) 3. Highly Skewed Data: Enable AQE Skew Join In heavy enterprise datasets, physical data is rarely organically distributed perfectly evenly. Imagine an active e-commerce platform where the default "Guest Customer" cleanly accounts for physically 60% of all universal platform transactions. If you intelligently execute a naive Sort-Merge Join broadly on customer_id, one single isolated Spark executor will physically be forced systematically to exclusively process the entire massive 60% "Guest" chunk. The other 199 regular executors will efficiently and cleanly finish in seconds and sit completely idle while that one node globally grinds to a halt. The rule: Actively, safely leverage Adaptive Query Execution (AQE) dynamically to natively, beautifully split heavily skewed partitions dynamically.The mechanism: AQE actively, dynamically, and securely detects massively skewed partitions directly mid-flight, accurately splitting them cleanly into optimally smaller, incredibly uniform, reliable sub-partitions so they can be effectively and seamlessly processed rapidly and cleanly in parallel. The Implementation Python # Ensuring AQE and Skew Join optimization are physically aggressively cleanly enabled locally in the specific cluster config spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true") The Silent Hero: Adaptive Query Execution (AQE) The part most data engineers fundamentally miss: AQE has actually been turned ON by default since Apache Spark 3.2. This means Spark is reading actual statistical data mid-job. If you execute a Sort-Merge Join on a massive table that unexpectedly shrinks to 40MB after an aggressive .filter() clause, AQE will actively intercept the job mid-flight and organically auto-switch the execution directly into a lightning-fast Broadcast Join. You technically don't have to code anything for this to happen. It organically just happens. But you do have to verify it. You must explicitly verify that spark.sql.adaptive.enabled is active in your environment. You must actively understand exactly what it is doing—because when AQE occasionally guesses wrong (usually due to stale table statistics), you need to know precisely how to aggressively override it with manual hints. Conclusion Performance tuning in distributed compute engines fundamentally comes down to actively understanding the physical network shuffle. Check your explicit joins. Aggressively read your physical query plans (using .explain()). And never blindly trust default configurations at the enterprise level. What is the absolute worst join performance bottleneck you have ever hit in production? Let me know in the comments below!

Change data capture (CDC) pipelines look straightforward on paper: capture database changes, publish them to Kafka, and update downstream systems. The difficulty starts when events are duplicated, consumers restart, projections drift, or a team needs to replay months of history without corrupting the state it is trying to recover. A reliable CDC design has to account for those failure modes from the beginning. That means combining Kafka and Debezium with idempotent writes, deterministic projections, controlled replay workflows, reconciliation checks, and enough recovery evidence to explain what happened when something goes wrong. The architecture: The goal is not only to move inventory changes quickly. The goal is to make replay safe enough that operators can rebuild and explain the derived state after failure. This article builds one concrete pattern: The important detail is that replay safety is not a single feature. It is the result of several boring decisions lining up correctly. Data Model The data model should separate the aggregate state, the classification state, and the transaction history. PLSQL CREATE TABLE inventory_stock_on_hand ( sku VARCHAR(64) PRIMARY KEY, stock_on_hand BIGINT NOT NULL, updated_at TIMESTAMP NOT NULL ); CREATE TABLE inventory_bucket ( sku VARCHAR(64) NOT NULL, bucket_type VARCHAR(32) NOT NULL, location_id VARCHAR(64) NOT NULL, quantity BIGINT NOT NULL, updated_at TIMESTAMP NOT NULL, PRIMARY KEY (sku, bucket_type, location_id) ); CREATE TABLE inventory_transaction ( event_id VARCHAR(128) PRIMARY KEY, sku VARCHAR(64) NOT NULL, seller_id VARCHAR(64) NOT NULL, delta_quantity BIGINT NOT NULL, event_time TIMESTAMP NOT NULL, accepted_at TIMESTAMP NOT NULL ); CREATE INDEX idx_inventory_transaction_sku_time ON inventory_transaction (sku, event_time); CREATE INDEX idx_inventory_bucket_sku_bucket ON inventory_bucket (sku, bucket_type); The transaction table is the recovery anchor. If the availability projection drifts, the system needs a history to explain the projection. Do not rely only on the mutable aggregate table. inventory_stock_on_hand is useful for fast reads, but it is not enough for recovery. If the aggregate is wrong, it cannot explain how it became wrong. The accepted transaction history gives replay something durable to reason from. Ingestion Event Use an event ID that can survive retries and replay. JSON { "event_id": "mkt-evt-8f11a", "sku": "1231241", "quantity": 100, "operation": "I", "event_time": "2026-06-19T18:23:11Z", "seller_id": "seller-42" } The consumer should perform an idempotent write. One pattern is to insert the transaction first using event_id as the primary key. If the insert fails because the event already exists, skip the duplicate and emit a duplicate-suppression metric. Java public InventoryWriteResult apply(InventoryEvent event) { try { transactionRepository.insert(event.toTransactionRow()); } catch (DuplicateKeyException duplicate) { metrics.increment("inventory.duplicate_event"); return InventoryWriteResult.duplicate(event.eventId()); } stockRepository.incrementStockOnHand(event.sku(), event.quantity()); bucketRepository.incrementBucket(event.sku(), "SELLABLE", event.quantity()); return InventoryWriteResult.accepted(event.eventId()); } In production, the accepted transaction insert and the aggregate updates should be part of the same database transaction. A useful shape is: PLSQL BEGIN; WITH accepted AS ( INSERT INTO inventory_transaction ( event_id, sku, seller_id, delta_quantity, event_time, accepted_at ) VALUES ( :event_id, :sku, :seller_id, :delta_quantity, :event_time, now() ) ON CONFLICT (event_id) DO NOTHING RETURNING sku, delta_quantity ) INSERT INTO inventory_stock_on_hand (sku, stock_on_hand, updated_at) SELECT sku, delta_quantity, now() FROM accepted ON CONFLICT (sku) DO UPDATE SET stock_on_hand = inventory_stock_on_hand.stock_on_hand + EXCLUDED.stock_on_hand, updated_at = now(); COMMIT; That ON CONFLICT clause is not just a database convenience. It is part of the replay contract. It ensures that retrying the same business event does not apply the same inventory delta twice. Debezium Configuration Enable PostgreSQL logical decoding and configure Debezium to emit CDC topics for the inventory tables. JSON { "name": "postgres-inventory-connector", "config": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "database.hostname": "<POSTGRES_HOSTNAME>", "database.port": "5432", "database.user": "<POSTGRES_USER>", "database.password": "<POSTGRES_PASSWORD>", "database.dbname": "<POSTGRES_DBNAME>", "topic.prefix": "inventory_source", "plugin.name": "pgoutput", "slot.name": "debezium_inventory_slot", "publication.autocreate.mode": "filtered", "table.include.list": "public.inventory_stock_on_hand,public.inventory_bucket,public.inventory_transaction", "snapshot.mode": "initial", "heartbeat.interval.ms": "10000", "tombstones.on.delete": "false", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable": "true", "value.converter.schemas.enable": "true" } } Debezium gives you history, but not recovery confidence. The confidence comes from how you key, project, replay, and reconcile that history. For replay work, track these connector facts in your runbook: Connector name and versionReplication slot namePublication name and included tablesSnapshot mode used for initial loadTopic prefixLast processed LSNConnector lagSchema history topic When a connector interruption happens, those details tell you whether you can resume normally, need a bounded replay, or need a new snapshot plus downstream reconciliation. Partition-Aware Routing The partition key should be chosen from the business ordering boundary. Java public class SkuPartitioner implements Partitioner { @Override public int partition( String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { InventoryEvent event = (InventoryEvent) value; String orderingKey = event.getSku(); int partitionCount = cluster.partitionCountForTopic(topic); return Math.floorMod(orderingKey.hashCode(), partitionCount); } } Partitioning is not merely a throughput setting. If the projection depends on entity-local ordering, the entity belongs in the key. Kafka Streams Topology A simplified topology might rekey CDC records by SKU, materialize source tables, and compute availability. Java StreamsBuilder builder = new StreamsBuilder(); KTable<String, StockOnHand> stock = builder.table("inventory_source.public.inventory_stock_on_hand", Consumed.with(Serdes.String(), stockSerde)); KTable<String, InventoryBuckets> buckets = builder.table("inventory_source.public.inventory_bucket", Consumed.with(Serdes.String(), bucketSerde)); KTable<String, AvailabilityProjection> availability = stock.join( buckets, (stockRow, bucketRows) -> AvailabilityProjection.compute(stockRow, bucketRows), Materialized.<String, AvailabilityProjection, KeyValueStore<Bytes, byte[]>>as("availability-store") .withKeySerde(Serdes.String()) .withValueSerde(availabilitySerde) ); availability .toStream() .filter((sku, projection) -> projection.isPublishable()) .to("inventory.availability.v2", Produced.with(Serdes.String(), availabilitySerde)); The projection function should be deterministic. If replaying the same accepted history does not produce the same projection, the topology is not replay-safe. Recovery Contract Attach a Recovery Contract to the flow. YAML recovery_contract: flow: inventory-availability-projection tuple: "<H, O, I, F, S, Q, E>" history: source: - inventory_transaction - debezium.inventory_transaction order: key: sku idempotency: key: event_id duplicate_policy: skip_and_report function: name: compute_sellable_availability deterministic: true scope: supported: - by_sku - by_time_window - by_partition checks: - stock_on_hand_matches_transactions - sellable_quantity_non_negative - projection_event_time_valid evidence: - replay_scope - events_processed - duplicates_skipped - projections_changed - reconciliation_failures - confidence_status Treat this file as executable architecture documentation. A service should fail fast if the contract is incomplete for a critical flow. Java public final class RecoveryContractValidator { public void validate(RecoveryContract contract) { requireNonEmpty(contract.flow(), "flow"); requireNonEmpty(contract.history().source(), "history.source"); requireNonEmpty(contract.order().key(), "order.key"); requireNonEmpty(contract.idempotency().key(), "idempotency.key"); requireNonEmpty(contract.function().name(), "function.name"); requireTrue(contract.function().deterministic(), "projection must be deterministic"); requireNonEmpty(contract.scope().supported(), "scope.supported"); requireNonEmpty(contract.checks(), "checks"); requireNonEmpty(contract.evidence(), "evidence"); } private void requireNonEmpty(Object value, String field) { if (value == null || value.toString().isBlank()) { throw new IllegalArgumentException("Missing recovery contract field: " + field); } } private void requireTrue(boolean value, String message) { if (!value) { throw new IllegalArgumentException(message); } } } That validator does not make the system correct by itself. It prevents a more common failure: discovering during an incident that nobody defined the replay scope, idempotency key, or reconciliation checks. Replay Workflow Replay should be treated as a controlled workflow. Plain Text 1. Identify incident scope. 2. Select replay scope by SKU, time window, or partition. 3. Read authoritative history. 4. Rebuild deterministic projection. 5. Run reconciliation checks. 6. Emit recovery evidence. 7. Republish only if checks pass. The output should be an evidence report. JSON { "recovery_id": "rec-2026-06-19-001", "flow": "inventory-availability-projection", "events_processed": 1842, "duplicates_skipped": 17, "projection_rows_changed": 11, "reconciliation": { "stock_on_hand_matches_transactions": true, "sellable_quantity_non_negative": true, "projection_event_time_valid": true }, "confidence_status": "trusted" } A replay runner can keep the workflow explicit: Java public RecoveryEvidence replay(ReplayRequest request) { RecoveryContract contract = contracts.load(request.flow()); validator.validate(contract); ReplayScope scope = scopeResolver.resolve(request, contract); List<InventoryEvent> history = historyReader.read(contract.history(), scope); ReplayResult result = projector.rebuild(history, contract.function()); ReconciliationResult reconciliation = reconciliationRunner.run(contract.checks(), scope, result); RecoveryEvidence evidence = RecoveryEvidence.builder() .recoveryId(UUID.randomUUID().toString()) .flow(request.flow()) .scope(scope) .eventsProcessed(history.size()) .duplicatesSkipped(result.duplicatesSkipped()) .projectionsChanged(result.changedRows()) .reconciliation(reconciliation) .confidenceStatus(reconciliation.passed() ? "trusted" : "review_required") .build(); evidenceStore.write(evidence); if (request.publish() && reconciliation.passed()) { publisher.publish(result.projections()); } return evidence; } The replay runner should support dry runs. Dry runs let operators answer "What would change?" before republishing availability, billing, or detection outputs. Operational Metrics Track ordinary health and recovery confidence separately. Ordinary health: Consumer lagConnector lagTask restartsDLQ countEnd-to-end latency Recovery confidence: Replay durationReplay scope sizeDuplicate suppression countProjection rows changedReconciliation failuresConfidence status Example metric names: Plain Text inventory_ingest_events_total{result="accepted|duplicate|rejected"} inventory_cdc_connector_lag_seconds{connector="postgres-inventory-connector"} inventory_stream_projection_lag_seconds{topology="availability"} inventory_replay_duration_seconds{flow="inventory-availability-projection"} inventory_replay_events_processed_total{flow="inventory-availability-projection"} inventory_replay_duplicates_skipped_total{flow="inventory-availability-projection"} inventory_reconciliation_failures_total{check="stock_on_hand_matches_transactions"} inventory_recovery_confidence_status{status="trusted|review_required|failed"} Alert on disagreement, not only lag. A good pipeline can be caught up and still be wrong. YAML alerts: - name: InventoryProjectionReconciliationFailure expr: inventory_reconciliation_failures_total > 0 severity: page - name: InventoryReplayRequiresReview expr: inventory_recovery_confidence_status{status="review_required"} > 0 severity: ticket - name: InventoryConnectorLagHigh expr: inventory_cdc_connector_lag_seconds > 300 severity: ticket Reconciliation Queries Reconciliation should be executable, not just a diagram in a runbook. Start with invariants that are simple enough to automate. Example: Stock-on-hand should match accepted transaction deltas for a replay window. PLSQL WITH accepted_delta AS ( SELECT sku, SUM(delta_quantity) AS expected_delta FROM inventory_transaction WHERE accepted_at BETWEEN :from_time AND :to_time GROUP BY sku ), actual_delta AS ( SELECT sku, stock_on_hand - :baseline_stock_on_hand AS observed_delta FROM inventory_stock_on_hand WHERE sku = :sku ) SELECT a.sku, a.expected_delta, b.observed_delta, (a.expected_delta = b.observed_delta) AS matches FROM accepted_delta a JOIN actual_delta b ON a.sku = b.sku; Example: Sellable inventory should never be negative. PLSQL SELECT sku, location_id, quantity FROM inventory_bucket WHERE bucket_type = 'SELLABLE' AND quantity < 0; These queries are not academically exciting, but they are operationally powerful. They turn "the replay finished" into "the replay finished and the invariants passed." Replay Endpoint Sketch A replay workflow should be explicit and permissioned. One possible internal API: HTTP POST /internal/recovery/replay Content-Type: application/json { "flow": "inventory-availability-projection", "scope": { "type": "sku_and_time_window", "sku": "1231241", "from_event_time": "2026-06-19T18:00:00Z", "to_event_time": "2026-06-19T19:00:00Z" }, "dry_run": false, "requested_by": "sre-oncall", "reason": "projection drift after stream task restart" } The response should not just say 200 OK. JSON { "recovery_id": "rec-2026-06-19-001", "status": "trusted", "events_processed": 1842, "duplicates_skipped": 17, "projections_changed": 11, "reconciliation_failures": 0, "evidence_uri": "<RECOVERY_EVIDENCE_URI>" } The response is the operational artifact. It gives the team something to attach to an incident timeline and something to compare against later recovery runs. Tests for Replay Safety Replay safety should be tested before production incidents. Java @Test void replayingSameHistoryDoesNotChangeProjectionTwice() { List<InventoryEvent> history = List.of( event("evt-1", "SKU-1", 10), event("evt-2", "SKU-1", -2), event("evt-1", "SKU-1", 10) // duplicate ); AvailabilityProjection first = projector.replay(history); AvailabilityProjection second = projector.replay(history); assertThat(first).isEqualTo(second); assertThat(first.sellableQuantity()).isEqualTo(8); assertThat(first.duplicatesSkipped()).isEqualTo(1); } Also test late events, schema versions, partition rebalance, connector restart, and partial replay by entity. If replay is part of your recovery model, it deserves the same test discipline as the happy-path pipeline. Add failure injection tests that mirror production recovery: Java @Test void lateEventTriggersReviewWhenItChangesPublishedAvailability() { ReplayScope scope = ReplayScope.forSkuAndWindow( "SKU-1", Instant.parse("2026-06-19T18:00:00Z"), Instant.parse("2026-06-19T19:00:00Z") ); history.append(event("evt-1", "SKU-1", 10, "2026-06-19T18:01:00Z")); history.append(event("evt-2", "SKU-1", -3, "2026-06-19T18:59:00Z")); history.appendLate(event("evt-3", "SKU-1", -2, "2026-06-19T18:30:00Z")); RecoveryEvidence evidence = replayRunner.replay( ReplayRequest.dryRun("inventory-availability-projection", scope) ); assertThat(evidence.eventsProcessed()).isEqualTo(3); assertThat(evidence.projectionsChanged()).isGreaterThan(0); assertThat(evidence.confidenceStatus()).isEqualTo("review_required"); } Failure Injection Matrix Use a small matrix before every major release of the pipeline. Duplicate Event Injection: Send the same event_id twice.Expected evidence: duplicates_skipped > 0; no double-counted stock.Late Event Injection: Delay event arrival until after the projection has already published output.Expected evidence: late event count, changed projections, and review status if the output changes.Connector Pause Injection: Stop the Debezium connector for several minutes.Expected evidence: connector lag, replay scope, and reconciliation status.Offset Rewind Injection: Reprocess a known event range.Expected evidence: deterministic replay agreement.Schema Change Injection: Replay old and new schema versions.Expected evidence: schema versions recorded in the recovery evidence.Bad projection deploy Injection: Publish an incorrect derived state, then replay.Expected evidence: projections changed; reconciliation passes after rebuild. The point is not to create chaos for its own sake. The point is to practice the exact recovery motion before a real incident. Production Hardening Checklist Before relying on replay in production, confirm: The authoritative history has retention longer than the largest expected recovery window.The idempotency key is stable across producer retries.The Kafka partition key matches the business ordering boundary.The projection function is deterministic for the supported replay scope.The contract names every source topic, source table, check, and evidence field.The replay endpoint supports dry runs.Republish requires reconciliation success.Evidence is written to durable storage.Evidence records include schema versions and replay input bounds.Operators can find the runbook from the alert.The DLQ is treated as an input to recovery, not as the recovery plan itself. For high-value flows, make this checklist part of the architecture review. It is much cheaper to define replay semantics while designing the pipeline than to invent them under pressure. Common Mistakes Treating CDC topics as transient integration messages instead of durable recovery history.Choosing partition keys for infrastructure convenience rather than business ordering.Allowing stream processors to perform hidden non-idempotent side effects.Measuring lag but not correctness.Resetting offsets without a reconciliation plan.Assuming exactly-once semantics removes the need for recovery evidence. Conclusion Replay-safe CDC pipelines require more than Kafka, Debezium, and stream processing. They require explicit recovery semantics. Recovery Contracts give teams a compact way to define those semantics. Confidence-carrying replay gives operators evidence that the recovered state can be trusted. That is the difference between a pipeline that resumes and a platform that actually recovers.

In a previous article, we built a static supply chain graph in Neo4j using Apache Spark, with suppliers, warehouses, distribution centers, and retailers connected by shipping routes. That gave us a snapshot of the network at a point in time. In this article, we'll add the streaming layer: shipment events flow through Confluent Cloud Kafka in real time, land in Neo4j as enriched graph properties, and a live dashboard shows network health updating as events arrive. The full source code is available on GitHub. The Stack Each tool in the stack does what it does best: ToolRoleConfluent Cloud (free tier)Managed Kafka cluster and topicPython producer (Jupyter)Generates and publishes synthetic shipment eventsPython consumer (Jupyter)Consumes events and writes them into Neo4jNeo4j AuraDBGraph database storing the supply chain and shipment eventsPlotlyLive dashboard visualization One deliberate omission is that we aren't using the Neo4j Kafka Sink Connector, which is available as a managed connector on Confluent Cloud. That connector handles the consumer side automatically but carries a per-task hourly charge. For this article, we'll keep everything free by writing a Python consumer that does the same job. This also has a practical benefit: all the pipeline logic is visible in Python rather than hidden inside a managed connector configuration, which makes it easier to understand and adapt. The managed connector is a natural next step for production workloads. Setting Up Confluent Cloud Sign up at confluent.io and create a free cluster.Once the cluster is running, create a topic named shipment-events with 1 partition and default settings.Create an API key and secret under API Keys.Note the bootstrap server address from the cluster settings. Export these as environment variables in your shell: Shell export CONFLUENT_BOOTSTRAP_SERVERS=your_cluster.confluent.cloud:9092 export CONFLUENT_API_KEY=your_api_key export CONFLUENT_API_SECRET=your_api_secret Setting Up Neo4j AuraDB AuraDB is Neo4j's fully managed cloud database. A free tier is available with no credit card required. Sign up at console.neo4j.io/graphacademy.Create a new AuraDB Free instance.When the instance is created, download or note the credentials — the connection URI, username, and password. Neo4j only shows the password once, so save it somewhere safe.Once the instance is running, open the built-in Query tab and verify connectivity: MATCH (n) RETURN count(n). This should return 0. We are ready to load data. Before starting Jupyter, export the connection details as environment variables in your shell: Shell export NEO4J_URI=neo4j+s://xxxx.databases.neo4j.io export NEO4J_USERNAME=your_username_here export NEO4J_PASSWORD=your_password_here export NEO4J_DATABASE=your_database_name_here The Data Model Each shipment event represents a single status update for a shipment at a point in time. A shipment does not generate a sequence of events as it progresses — each event is an independent snapshot, which keeps the producer simple and the consumer stateless. The event structure is: JSON { "shipment_id": "c60eb761-f153-4840-8427-17fa9e34c56c", "supplier_id": "S013", "warehouse_id": "W005", "dist_center_id": "DC004", "retailer_id": "R025", "status": "delayed", "timestamp": "2026-08-04T12:57:15Z", "delay_minutes": 34 } Status follows one of four values — departed, in_transit, delayed or delivered, with a configurable delay probability. We use 15% delayed to make the dashboard interesting without overwhelming it. When the consumer writes an event into Neo4j, it creates a Shipment node and links it to the existing supply chain nodes via four relationship types: Cypher MERGE (sh:Shipment {shipment_id: $shipment_id}) SET sh.status = $status, sh.timestamp = $timestamp, sh.delay_minutes = $delay_minutes WITH sh MATCH (s:Supplier {id: $supplier_id}) MATCH (w:Warehouse {id: $warehouse_id}) MATCH (dc:DistributionCenter {id: $dist_center_id}) MATCH (r:Retailer {id: $retailer_id}) MERGE (s)-[:HAS_SHIPMENT]->(sh) MERGE (sh)-[:VIA_WAREHOUSE]->(w) MERGE (sh)-[:VIA_DIST_CENTER]->(dc) MERGE (sh)-[:DESTINED_FOR]->(r) MERGE on shipment_id means re-running the consumer never creates duplicate nodes. The Producer The producer notebook uses a fixed random seed to generate reproducible shipment events using IDs drawn from the existing supply chain and publishes them to Confluent Cloud via the confluent-kafka library: Python producer = Producer({ "bootstrap.servers": BOOTSTRAP_SERVERS, "security.protocol": "SASL_SSL", "sasl.mechanisms": "PLAIN", "sasl.username": API_KEY, "sasl.password": API_SECRET, "log_level": 0, }) Setting "log_level": 0 suppresses the librdkafka telemetry messages that appear otherwise. The producer supports both batch and continuous modes. For example: Python produce_events(num_events = -1) # stream continuously produce_events(num_events = 100) # publish exactly 100 events The display refreshes every PRINT_EVERY events using clear_output, showing the latest event and a running status breakdown — so the cell output stays manageable even when streaming thousands of events. The Consumer and Live Dashboard Rather than two separate notebooks, we combine the consumer and dashboard into a single pipeline. On each cycle, the loop: Polls Kafka for up to POLL_BATCH events and writes them to Neo4jQueries Neo4j for the current graph stateRebuilds and redraws the dashboardSleeps for REFRESH_INTERVAL seconds before repeating Rebuilding the full dashboard on every cycle is straightforward and works well at demo event rates. At higher throughput, a more efficient approach would be to update only the changed data rather than redrawing all eight panels on each refresh. The consumer uses its own Kafka group ID (supply-chain-dashboard) so it reads the topic independently, catching up on all existing events first before staying live: Python consumer = Consumer({ "bootstrap.servers": BOOTSTRAP_SERVERS, "security.protocol": "SASL_SSL", "sasl.mechanisms": "PLAIN", "sasl.username": API_KEY, "sasl.password": API_SECRET, "group.id": "supply-chain-dashboard", "auto.offset.reset": "earliest", "log_level": 0, }) The Live Dashboard The dashboard uses Plotly's make_subplots in a 4x2 grid, rebuilt on every refresh cycle using clear_output. Eight panels give a complete picture of network health: Row 1 – Overall Health Network status table: Total shipments, delayed count, delay rate, Kafka events consumed, refresh count, and any disabled nodesShipment status distribution: Donut chart showing the split between departed, in transit, delayed, and delivered, as shown in Figure 1 Figure 1. Shipment Status Distribution Row 2 – Warehouse View Delayed shipments by warehouse: Which warehouses are handling the most delayed shipments right nowWarehouse health score: A heatmap scoring each warehouse from 0.0 (everything delayed) to 1.0 (fully healthy), colored red through orange to green, as shown in Figure 2 Figure 2. Warehouse Health Score Row 3 – Origin and Destination Supplier performance: Which suppliers are generating the most delayed shipmentsRetailer impact: Which retailers are receiving the most delayed shipments — the downstream effect of any disruption Row 4 – Mid-Network and Flow Average delay by distribution center: Where in the middle layer delays are accumulatingShipment flow: A Sankey diagram (Figure 3) showing which suppliers are routing through which warehouses Figure 3. Shipment Flow - Suppliers to Warehouses The warehouse health score is the most immediately readable panel. The Cypher behind it computes the score directly in the graph: Cypher MATCH (sh:Shipment)-[:VIA_WAREHOUSE]->(w:Warehouse) WHERE w.active IS NULL OR w.active <> false WITH w.id AS warehouse, count(sh) AS total, count(CASE WHEN sh.status = 'delayed' THEN 1 END) AS delayed RETURN warehouse, round(1.0 - toFloat(delayed) / total, 3) AS health_score ORDER BY warehouse Simulating a Network Disruption One of the more compelling features of the graph model is how easy it is to simulate and visualize a disruption. Setting active = false on any node excludes it from the dashboard queries and the dashboard immediately reflects the simulated disruption on the next refresh cycle. We can do this before the dashboard starts: Python REMOVE_NODE = "W007" # mark this warehouse as inactive Or live, while the dashboard is running, using the Neo4j AuraDB Query tab: Cypher // Disable a node MATCH (n {id: "W007"}) SET n.active = false // Re-enable a node MATCH (n {id: "W007"}) REMOVE n.active // Check what is currently disabled MATCH (n) WHERE n.active = false RETURN labels(n)[0] AS label, n.id AS id Within 5 seconds, the dashboard reflects the change. The warehouse health heatmap shows the gap, the delayed shipments bar shifts to other warehouses as traffic reroutes, and the network status table shows the node as disabled. Re-enabling it and watching the metrics recover completes the disruption and recovery story. Standalone Operation At startup, the consumer notebook creates the supply chain nodes using MERGE. This operation is idempotent, so any existing nodes from the previous article are left unchanged. Note that this step creates nodes only — the relationships between supply chain nodes (supplier -> warehouse -> distribution center -> retailer) are assumed to exist from the previous article, or can be added separately if running this notebook in isolation. Python with driver.session(database = NEO4J_DATABASE) as session: for i in range(20): session.run("MERGE (:Supplier {id: $id})", id = f"S{i:03d}") for i in range(12): session.run("MERGE (:Warehouse {id: $id})", id = f"W{i:03d}") for i in range(10): session.run("MERGE (:DistributionCenter {id: $id})", id = f"DC{i:03d}") for i in range(30): session.run("MERGE (:Retailer {id: $id})", id = f"R{i:03d}") Gotchas and Lessons Learned Suppress librdkafka Logging Without "log_level": 0 in the producer and consumer config, Confluent's underlying librdkafka library prints telemetry messages to the cell output every time a connection is established. The messages are harmless. Suppress Neo4j Property Warnings Querying a property that does not yet exist on any node produces a GqlStatusObject warning from Neo4j for every query that references it. The active property falls into this category when no node has been disabled. The fix is one line to set notifications to "OFF" on the driver, as follows: Python driver = GraphDatabase.driver( NEO4J_URI, auth = (NEO4J_USERNAME, NEO4J_PASSWORD), notifications_min_severity = "OFF", ) Consumer Group Isolation Kafka distributes partitions across consumers in the same group, so each consumer processes only its assigned partitions. If we run multiple consumers using the same group ID against the same topic, each will only process a subset of the events. The dashboard uses supply-chain-dashboard as its group ID, and the tip is to run only one instance of this notebook at a time against the same topic and cluster. auto.offset.reset = earliest Without this setting, a consumer that starts after events have been published will miss everything that arrived before it connected. Setting earliest means the consumer always catches up on the full history of the topic before going live, which is essential if we stop and restart the dashboard mid-session. Clear Shipment Nodes Between Runs Each run of the consumer creates new Shipment nodes. Since the producer generates synthetic demo data, it's safe to clear these between runs; otherwise, successive runs would accumulate all historical shipments, and the dashboard counts would grow unbounded. The notebook clears all Shipment nodes at startup: Cypher MATCH (sh:Shipment) CALL (sh) { DETACH DELETE sh } IN TRANSACTIONS OF 10000 ROWS Summary We've built a real-time supply chain event streaming pipeline using Confluent Cloud Kafka and Neo4j. The producer generates synthetic shipment events continuously, the consumer writes them into the graph, and a live dashboard shows network health updating in near real-time. The disruption simulation — marking a node inactive mid-run and watching the dashboard respond — demonstrates one of the most compelling aspects of the graph model: the ability to ask structural questions about a network as it evolves. The same architecture adapts naturally to real logistics, IoT, or manufacturing event streams where understanding network structure matters as much as raw throughput. The full source code is available on GitHub.

By Akmal Chaudhri CORE

Reliable orchestration for small language models depends less on model sophistication than on the durability of event flow and state. Under the assumptions used here — small model instances, little or no local state, Kafka as the event backbone, Temporal as the orchestration layer and durable state store, and Java as the runtime — the safest design is to treat model invocations as replayable side effects, Kafka as the transport and ordering substrate, and Temporal Workflow state as the canonical record of conversational progress. In that design, Kafka provides high-throughput append-only event delivery and partition-local ordering, while Temporal persists Workflow Event History and can replay execution after failures. Exactly-once semantics remain meaningful inside Kafka’s consume-transform-produce boundary when transactions and read_committed are used, but once processing crosses into external systems such as model APIs, durable activities, or databases, correctness comes from idempotency, deduplication, sequence checks, and reconciliation rather than from a global exactly-once guarantee. Assumptions The most productive baseline is a narrow one. Each conversation, task, or model session is keyed so related events land on the same Kafka partition, preserving order only where order actually exists: within one partition, not across the topic. Each workflow instance owns one conversational state machine, stores the minimal context needed to decide the next action, and invokes model calls through Temporal Activities so failures, retries, and timeouts are visible and durable. Large prompts, attachments, or long transcripts are not kept as incidental JVM memory because Temporal persists inputs and outputs in Event History and large histories degrade replay latency; those artifacts belong in external storage with durable references held in workflow state. Analysis The central engineering mistake in LLM orchestration is to confuse transport delivery with business completion. Kafka can guarantee at-least-once delivery by processing records before committing consumer offsets, and it can provide exactly-once behavior for Kafka-to-Kafka pipelines by atomically updating produced records and consumed offsets with transactions. Kafka’s own design documentation is explicit that the producer is the transactional component and that read_committed is advisable when aiming for exactly-once processing. The same documentation also makes clear why the guarantee weakens at system boundaries: once consumed data must be coordinated with an external state store or side effect, the problem becomes cross-system consistency rather than log delivery. In a Temporal-based model pipeline, that means Kafka should usually be treated as the durable ingress path, while Temporal owns the authoritative notion of whether an event was applied to a conversation state machine. That separation suggests a simple rule. Offsets are transport progress; workflow state is semantic progress. A consumer should therefore commit offsets only after handoff to a durable semantic owner. In this architecture, that owner is the Temporal workflow receiving a signal. Temporal workflows behave like stateful services that receive Signals, Queries, and Updates, and the platform persists Event History so a crashed worker can replay the workflow and resume from the last recorded event. Signal handlers are allowed to mutate workflow state, and blocking coordination can be expressed safely with Workflow.await. Activity retries are configured through ActivityOptions and RetryOptions, with heartbeat support for long-running calls. Java @WorkflowInterface interface ModelFlow { @WorkflowMethod void run(String sessionId); @SignalMethod void onEvent(ModelEvent event); @QueryMethod long lastAppliedSequence(); } private final ModelActivities activities = Workflow.newActivityStub( ModelActivities.class, ActivityOptions.newBuilder() .setStartToCloseTimeout(Duration.ofSeconds(20)) .setRetryOptions( RetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(250)) .setMaximumAttempts(5) .build()) .build()); private final NavigableMap<Long, ModelEvent> pending = new TreeMap<>(); private long nextSequence = 1; private ConversationState state = ConversationState.empty(); @Override public void onEvent(ModelEvent event) { pending.putIfAbsent(event.sequence(), event); } @Override public void run(String sessionId) { for (;;) { Workflow.await(() -> pending.containsKey(nextSequence) || state.closed()); if (state.closed()) break; var event = pending.remove(nextSequence); state = activities.applyEvent(state, event); nextSequence = event.sequence() + 1; } Workflow.await(Workflow::isEveryHandlerFinished); } @Override public long lastAppliedSequence() { return nextSequence - 1; } This workflow fragment does three important things at once. The @SignalMethod declares asynchronous event ingress, the @QueryMethod exposes durable progress for reconciliation, and the activity stub attaches retry policy directly to the state transition that may call a model endpoint or another dependency. The pending map is not a queue for throughput; it is a reordering guard. If Kafka redelivers a message or an upstream retry arrives out of sequence, putIfAbsent and the nextSequence gate prevent semantic duplication and preserve per-session causality. Finishing the run only after Workflow.isEveryHandlerFinished() avoids the Temporal-documented failure mode where a workflow completes or continues-as-new while a handler is still waiting on asynchronous work. The matching Kafka consumer must be deliberately conservative. Automatic commits are inappropriate because they advance transport progress in the background regardless of semantic application. Manual synchronous commits make the boundary explicit, and Kafka documents that committed offsets are the secure restart position, and that commitSync should write the next offset, meaning lastProcessedOffset + 1. The consumer is also not thread-safe, so per-partition in-order handling is easiest when one poll loop owns one consumer instance and performs durable handoff before commit. Java void pollLoop() { consumer.subscribe(List.of("model-events")); while (running.get()) { var records = consumer.poll(Duration.ofSeconds(1)); for (var partition : records.partitions()) { var batch = records.records(partition); for (var record : batch) { var eventId = header(record, "event-id"); if (!inbox.tryInsert(eventId, record.topic(), record.partition(), record.offset())) { continue; } var workflow = client.newWorkflowStub(ModelFlow.class, record.key()); workflow.onEvent(ModelEvent.from(record)); inbox.markApplied(eventId); } var nextOffset = batch.get(batch.size() - 1).offset() + 1; consumer.commitSync(Map.of(partition, new OffsetAndMetadata(nextOffset))); } if (inbox.backlog() > 50_000) consumer.pause(consumer.assignment()); else consumer.resume(consumer.assignment()); } } The durable inbox is the effective-once bridge. If the process crashes after signaling Temporal but before committing offsets, Kafka may redeliver, yet tryInsert suppresses reapplication. If upstream producers use Kafka transactions, the consumer should read with isolation.level=read_committed so aborted records stay invisible; Kafka’s configuration reference notes that read_committed returns only committed transactional messages and withholds records past the last stable offset while open transactions exist. Backpressure also belongs here. Kafka exposes pause and resume without forcing a group rebalance, and monitoring guidance explicitly recommends watching lag, fetch rate, poll timing, and commit latency to ensure consumers are keeping up. Context propagation is easiest when context is split into stable metadata and mutable conversational state. Stable identifiers such as trace ID, tenant, policy version, and conversation key belong in Kafka headers and Temporal headers so they survive hops across services and activities; Kafka’s ProducerRecord supports headers, and Temporal context propagators move custom key-value data across workflow, activity, and child-workflow boundaries. Mutable context, by contrast, should not live in worker memory or ad hoc caches. It belongs in the workflow state, often as a compact summary plus references to offloaded artifacts. Temporal’s documentation explicitly warns that all activity inputs and outputs are persisted, that long AI-style conversations grow history, and that large histories degrade workflow-task latency. For long-running sessions, Continue-As-New provides a checkpoint boundary with a fresh Event History while preserving the workflow identity chain. Reconciliation closes the last reliability gap. Even with careful commits, outages, manual replays, or producer bugs can create suspicion that a workflow missed an event. Temporal queries are read-only and must not mutate state or block, which makes them ideal for asking a workflow for its durable high-water mark and replaying any gap from the event store. Java void reconcile(String workflowId, long durableHighWatermark) { var workflow = client.newWorkflowStub(ModelFlow.class, workflowId); long applied = workflow.lastAppliedSequence(); eventStore.readRange(workflowId, applied + 1, durableHighWatermark) .forEach(workflow::onEvent); } This pattern works because the workflow does not trust delivery history alone; it trusts its own durable state. Observability then becomes the enforcement layer for those guarantees. Kafka should surface lag, request latency, retry rates, poll gaps, and buffer exhaustion, while Temporal should emit metrics through Micrometer, trace activity and workflow paths, and expose searchable workflow metadata through Search Attributes. Temporal also recommends monitoring replay latency because large histories, payload sizes, and cache churn drive recovery cost. Together, these signals reveal the difference between a system that is slow, a system that is duplicating work, and a system that is actually losing context. Conclusion Orchestrating small language models without losing events or context is fundamentally a durability problem, not a prompt-engineering problem. Kafka should be used for ordered transport and scalable ingestion, but semantic completion should be anchored in Temporal’s durable workflow state, where signals, sequence gates, retryable activities, queries, and replay make failures recoverable rather than ambiguous. Exactly-once remains valuable inside Kafka’s transactional envelope, yet end-to-end correctness across model calls and other side effects comes from explicit idempotency, deduplication, reconciliation, and bounded context management with external storage and continue-as-new. In a Java stack, that combination yields an architecture where duplicates become harmless, ordering becomes explicit, back pressure becomes controlled, and context survives crashes because it is recorded in durable state instead of being left in process memory.

By Akhil Madineni CORE

I work as a data analyst at a legal services company. Part of my work involves protecting sensitive data during the Test Data Management (TDM) process. Many other departments in the company need test data to develop an application. Copying the production data for test sounds like a good plan. But because the test environment usually has lower cybersecurity requirements, this will cause customer privacy data leaks. So, my job is to mask the sensitive data to protect customer privacy. When it comes to my job, the first thing that comes to many people’s minds is that my work involves masking sensitive data. For example, changing the email address from [email protected] to [email protected]. Masking data is indeed important, but before we jump to the masking step, there's one basic question: Which column contains sensitive data, and how can I find it? In this article, I will introduce a pipeline designed to identify sensitive data columns before masking steps. Structure of the Pipeline Please find the identifying sensitive data pipeline structure workflow chart below: Identifying Sensitive Data Pipeline Structure Workflow Before I start introducing each stage, I’d like to mention two points. The first point: The original intent behind this pipeline structure design was to save time spent locating sensitive data. Server usage is billed based on duration. In a perfect world, the system would balance efficiency and accuracy. However, in practice, efficiency takes precedence in order to cut costs. The second point: The pipeline also had to preserve data usability for testing. In some cases, data privacy controls must be designed in a way that does not break core application workflows. For key columns such as primary keys and foreign keys, they need to preserve join functions and application workflows. So in practice, we usually leave them unchanged. Apply Column Name and Pattern Matching First, and quite intuitively, many columns' names are really straightforward and can be easily identified. For example, full name, phone number, and email. After the very first easy screening, some columns can be identified by hardcoded Python scripts, based on the specific column name pattern. However, there is an issue at this stage. I can identify columns containing sensitive data using customer email. But if there is another column named customer email address that hasn't been included in the hardcoded script, I won't be able to detect it. Besides that, relying solely on column names isn't always reliable. Take the notes column, a free-text field, for instance. It often appears as an optional field after the main information has been entered. Most people will leave it blank or write some insignificant things. But sometimes customers do write something, such as Our CEO Everett would like you to prioritize processing the ABC document. Please email them to [email protected] as soon as you finish, and then call 123-456-7890 to notify him. If I don't mark this column as need masking, the customers' private information will be exposed. Check Historical Decisions After the initial filtering step, I will check the historical decisions database for specific columns, such as the notes column mentioned earlier. If the database indicates that the historical decision for column notes is to mask it, then that column will be masked during the current round. Even if the notes in this specific round contain no sensitive data. For example, no privacy-related information is mentioned. There is no guarantee that the data in the next refreshed cycle will remain free of sensitive information. Send Ambiguous Columns to AI for Review and Analyze Sample Values Here comes the highlight of the entire pipeline. Sometimes, column names are somewhat ambiguous. Or it's unclear whether certain rows contain sensitive data. Let's take the notes column mentioned earlier again. It might be empty. Or it could contain a message like When food is delivered, please ring the doorbell and call my wife Bobi, thereby the sensitive information gets leaked. I started using the spaCy library from Python for Natural Language Processing (I will refer to this term as NLP later in this article). While spaCy isn’t a Large Language Model (I will call this term LLM), it certainly performs NLP analysis. However, the sampling process was time-consuming. I would sample the entire dataset if it had fewer than 50,000 rows, but randomly select 50,000 rows if it exceeded that limit. In a later version of the workflow, I switched to OpenAI: this time, I just need to select a sample of 100 rows and send them via API to the AI/LLM for analysis. The AI then generates a masking recommendations database, which will undergo manual review later. Accuracy improved significantly after we began using LLMs. It rose from 80% with spaCy to approximately 93% after switching to OpenAI. This 93% figure was determined by having human analysts conduct a column-by-column analysis in parallel with my development of the pipeline and automation scripts. So the result is benchmarked against manual reviews. Furthermore, this figure represents an average obtained after two rounds of actual TDM data masking operations and several additional rounds of testing. Regarding the remaining 7% of errors, false positives accounted for about 90%, and false negatives for only 10%. This is important because missing sensitive data is much more serious than over-flagging a column for review. Compared to manually analyzing a medium-sized schema containing 100 tables for 64 hours. An automated script can complete the analysis in just 2 hours. However, please note that this 2-hour timeframe does not include the time required for subsequent manual review. Human Review and Store Recommendations and New Decisions After the AI/LLM finishes analysis, human analysts will review the mask recommendations database generated by the AI. Each row in the database generates a report containing the user ID, database name, table name, column name, masking suggestion, masking rule, and analysis date. Then, humans will review the mask suggestions and corresponding masking methods. For example, the AI-generated mask suggestion database is: AI-generated Mask Suggestion Database Example As a human analyst, at this stage, I can review the masking suggestion generated by the AI. I would agree with the suggestion to mask the data. However, regarding the masking rule, I would review it and change it to set it to a blank value. Manual review needs to randomly sample 500 rows and analyze them individually to reach a final mask decision. In this new process, human analysts only need to review a single row of AI-generated mask decisions and mask rules. The switch saves time significantly. During a new round of the TDM data masking process, some new columns will be identified by AI and flagged as requiring masking. The new masking decision will be added to the existing historical decisions database after manual review. Send to Data Governance and Send to Business Customer and Get Feedback After our TDM team identifies and masks the sensitive data columns, we submit our results to the Data Governance department for a secondary manual review. Their review process differs slightly from ours. Our team focuses on using business knowledge to determine whether a column contains sensitive data. And we’re also responsible for developing more efficient identification & masking procedures. However, the Data Governance department needs to review and provide more accurate masking decisions. Because their team members have better knowledge of how to decide whether a column should be masked and of the appropriate masking method. After our two departments conducted two rounds of manual review, we sent the masked data results to our business customers' departments. They will use this data for testing and provide us with feedback based on their specific needs. For example, we recommended masking customer_id with a generated synthetic number. But doing so will change primary and foreign keys, thereby breaking database linkages. So, our business customer departments advised us against masking those columns. Conclusion and Future Improvement Plans Successfully masking sensitive data begins with accurately identifying the columns containing such data. Many people skip this and jump straight to the more interesting masking process. In my view, however, getting this step wrong will fail the rest of the workflow as well. The pipeline I designed isn't perfect. And I have a few ideas for improving the "Apply column name and pattern matching" component in the future. Since we’ve already used OpenAI, why not let the AI detect new patterns when analyzing ambiguous columns? We could have the AI generate a dynamic pattern database that updates automatically with every refresh cycle. It would also help us continuously update and refine our historical decisions database.

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

By Janani Annur Thiruvengadam CORE

Why Delta Lake? Apache Parquet on cloud storage was a great first step for data lakes — but it left engineers dealing with a painful set of problems in production: No ACID transactions — concurrent reads/writes could corrupt data silentlySchema drift — nothing stopped upstream systems from changing column typesNo deletes or updates — GDPR compliance meant rewriting entire partitionsPainful failure recovery — half-written data after a job crash became your problem Delta Lake solves all of this by sitting on top of Parquet and adding a transaction log (_delta_log/) that records every operation atomically. On Databricks, Delta is the default table format, deeply integrated with Apache Spark, Auto Optimize, and the Photon execution engine. The Medallion Architecture The medallion (or multi-hop) architecture organizes data into three progressive refinement layers. Each layer has a clear contract with the layers around it. Each layer has a distinct responsibility: LayerAliasPurposeRetentionBronzeRawLand data as-is, preserve source fidelityYears (audit trail)SilverCleansedDeduplicate, validate, type-cast, conform schemasMonthsGoldAggregatedBusiness-level KPIs, domain-specific aggregatesMonths–Years The key design principle is: never skip a layer. Debugging a production incident is infinitely easier when you can replay from raw Bronze data. Delta Lake Internals: The Transaction Log Before writing a single line of code, it's worth understanding how Delta Lake achieves ACID semantics. Every write operation (INSERT, UPDATE, DELETE, MERGE) produces a new JSON entry in _delta_log/. This log-based approach gives you time travel for free — querying VERSION AS OF 2 simply replays only the log entries up to that point. Checkpoints every 10 commits keep read performance snappy even with thousands of versions. Setting Up Your Databricks Environment All code here runs on Databricks Runtime 13.x+ (which ships Delta Lake 2.x). For local dev, use the delta-spark package. # Databricks notebook — Runtime 13.3 LTS or higher # Delta Lake is pre-installed; no pip install needed from pyspark.sql import SparkSession from pyspark.sql import functions as F from pyspark.sql.types import ( StructType, StructField, StringType, LongType, DoubleType, TimestampType, BooleanType ) from delta.tables import DeltaTable # On Databricks, SparkSession is pre-created as `spark` # For local testing: # spark = (SparkSession.builder # .appName("medallion-pipeline") # .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") # .config("spark.sql.catalog.spark_catalog", "spark.sql.delta.catalog.DeltaCatalog") # .getOrCreate()) # Unity Catalog paths (recommended for production) CATALOG = "prod" BRONZE_DB = f"{CATALOG}.bronze" SILVER_DB = f"{CATALOG}.silver" GOLD_DB = f"{CATALOG}.gold" # DBFS / external storage path (for raw file landing) RAW_LANDING = "abfss://[email protected]/events/" Building the Bronze Layer Bronze is your append-only raw ingestion layer. The goal is landing data with zero transformation — preserve everything, even malformed records. Schema is inferred or declared loosely. # ── Bronze Ingestion ────────────────────────────────────────────────────────── # Using Auto Loader (cloudFiles) — the Databricks-native incremental ingest # mechanism. It tracks which files have been processed via a checkpoint, # so re-running never double-injects data. raw_event_schema = StructType([ StructField("event_id", StringType(), True), StructField("user_id", StringType(), True), StructField("event_type", StringType(), True), StructField("event_ts", StringType(), True), # keep as string in bronze StructField("properties", StringType(), True), # raw JSON blob StructField("session_id", StringType(), True), StructField("platform", StringType(), True), ]) bronze_stream = ( spark.readStream .format("cloudFiles") .option("cloudFiles.format", "json") .option("cloudFiles.schemaLocation", "/checkpoints/bronze_events_schema") .option("cloudFiles.inferColumnTypes", "false") # keep raw types .schema(raw_event_schema) .load(RAW_LANDING) # Enrich with ingestion metadata — critical for debugging .withColumn("_ingest_ts", F.current_timestamp()) .withColumn("_source_file", F.input_file_name()) .withColumn("_ingest_date", F.to_date(F.current_timestamp())) ) ( bronze_stream.writeStream .format("delta") .outputMode("append") .option("checkpointLocation", "/checkpoints/bronze_events") .option("mergeSchema", "true") # allow new columns from upstream .partitionBy("_ingest_date") # partition for incremental silver reads .trigger(availableNow=True) # run-once trigger (for scheduled jobs) .tableCheckpoint(f"{BRONZE_DB}.events_raw") .toTable(f"{BRONZE_DB}.events_raw") ) Pro tip: Always add _ingest_ts and _source_file metadata columns in Bronze. When an upstream system sends corrupt data at 3 AM, these columns tell you exactly which file batch caused it. Transforming to Silver Silver is where the real engineering work happens: deduplication, type casting, schema validation, and applying business rules. We also handle CDC (Change Data Capture) upserts here using Delta's MERGE. # ── Silver Transformation ───────────────────────────────────────────────────── def transform_bronze_to_silver(bronze_df): """ Apply cleansing and conforming rules to raw Bronze events. Returns a Silver-ready DataFrame with enforced schema. """ return ( bronze_df # 1. Parse timestamps properly .withColumn("event_ts", F.to_timestamp("event_ts", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")) # 2. Parse the raw JSON properties blob into a struct .withColumn("props", F.from_json( F.col("properties"), schema="page_url STRING, referrer STRING, duration_ms LONG, revenue DOUBLE" )) .drop("properties") # 3. Normalize platform values .withColumn("platform", F.lower(F.trim(F.col("platform")))) .withColumn("platform", F.when( F.col("platform").isin("ios", "android"), F.col("platform") ).when( F.col("platform").isin("web", "browser", "desktop"), F.lit("web") ).otherwise(F.lit("unknown"))) # 4. Filter out test/internal traffic .filter(~F.col("user_id").startswith("test_")) .filter(F.col("event_id").isNotNull()) # 5. Deduplicate within the micro-batch (window by event_id) .dropDuplicates(["event_id"]) # 6. Add Silver metadata .withColumn("_silver_ts", F.current_timestamp()) .withColumn("event_date", F.to_date("event_ts")) # partition key ) # Upsert into Silver using MERGE (handles late-arriving / duplicate events) def upsert_to_silver(micro_batch_df, batch_id): micro_batch_df = transform_bronze_to_silver(micro_batch_df) silver_table = DeltaTable.forName(spark, f"{SILVER_DB}.events_clean") ( silver_table.alias("target") .merge( micro_batch_df.alias("source"), "target.event_id = source.event_id" # dedup key ) .whenMatchedUpdateAll() # update if record arrived late with corrections .whenNotMatchedInsertAll() # insert new records .execute() ) # Stream from Bronze → Silver using foreachBatch ( spark.readStream .format("delta") .option("readChangeFeed", "true") # CDF — only process new Bronze rows .table(f"{BRONZE_DB}.events_raw") .writeStream .foreachBatch(upsert_to_silver) .option("checkpointLocation", "/checkpoints/silver_events") .trigger(availableNow=True) .start() ) Aggregating to Gold Gold tables are business-ready aggregates consumed directly by BI tools, dashboards, and ML feature pipelines. They are typically batch-refreshed on a schedule. # ── Gold Aggregation ────────────────────────────────────────────────────────── daily_revenue = ( spark.table(f"{SILVER_DB}.events_clean") .filter(F.col("event_type") == "purchase") .filter(F.col("event_date") >= F.date_sub(F.current_date(), 90)) # rolling 90d .groupBy("event_date", "platform") .agg( F.sum("props.revenue").alias("total_revenue"), F.countDistinct("user_id").alias("unique_buyers"), F.count("event_id").alias("transaction_count"), F.avg("props.duration_ms").alias("avg_session_duration_ms"), ) .withColumn("revenue_per_buyer", F.round(F.col("total_revenue") / F.col("unique_buyers"), 2)) .withColumn("_gold_ts", F.current_timestamp()) ) # Overwrite with replaceWhere — only touch the last 90 days, not the full table ( daily_revenue.write .format("delta") .mode("overwrite") .option("replaceWhere", "event_date >= date_sub(current_date(), 90)") .saveAsTable(f"{GOLD_DB}.daily_revenue") ) Z-Ordering and Data Skipping Z-ordering is Databricks' multi-dimensional clustering technique. It co-locates related data within the same set of Parquet files, so Spark can skip irrelevant files entirely during queries — without the overhead of strict partitioning. -- Run OPTIMIZE + ZORDER after significant writes -- This rewrites data files to cluster on the most commonly filtered columns OPTIMIZE prod.silver.events_clean ZORDER BY (user_id, event_date, event_type); -- Check how many files were skipped in your last query -- (run immediately after a SELECT with filters) SELECT operation, operationMetrics['numFilesAdded'] AS files_added, operationMetrics['numFilesRemoved'] AS files_removed, operationMetrics['numRemovedBytes'] AS bytes_removed FROM ( DESCRIBE HISTORY prod.silver.events_clean ) WHERE operation = 'OPTIMIZE' ORDER BY timestamp DESC LIMIT 5; Rule of thumb: Z-order on your top 3–4 most-filtered columns. Beyond that, the clustering benefit diminishes and OPTIMIZE runtimes grow significantly. Never Z-order on partition columns — they're already physically separated. Optimized Spark Writes Poorly tuned Spark writes are the #1 cause of small-file problems in Delta Lake. Here's a production-hardened write configuration: # ── Write Configuration Reference ──────────────────────────────────────────── SILVER_WRITE_CONFIG = { # Coalesce output files to ~128MB each (avoids small-file explosion) "spark.sql.shuffle.partitions": "200", # tune to cluster size "spark.databricks.delta.optimizeWrite.enabled": "true", # auto bin-packing "spark.databricks.delta.autoCompact.enabled": "true", # background compaction # Target file size for Auto Optimize "spark.databricks.delta.optimizeWrite.binSize": "134217728", # 128 MB in bytes # Enable deletion vectors (Databricks 12.2+) — soft-deletes without file rewrites "spark.databricks.delta.enableDeletionVectors": "true", } # Apply at the session level for the pipeline job for k, v in SILVER_WRITE_CONFIG.items(): spark.conf.set(k, v) # For partitioned tables: control output file count per partition ( silver_df .repartition(F.col("event_date")) # one task group per date partition .write .format("delta") .mode("overwrite") .option("dataChange", "true") .option("overwriteSchema", "false") # never silently change schema in prod .partitionBy("event_date") .saveAsTable(f"{SILVER_DB}.events_clean") ) Pipeline Comparison Table Here's how different ingestion patterns stack up for common production scenarios on Databricks: PatternLatencyThroughputDedup SupportBest ForAuto Loader + AppendNear-real-timeVery High❌ NoEvent logs, immutable streamsAuto Loader + MERGENear-real-timeHigh✅ YesCDC, late-arriving eventsBatch COPY INTOMinutesHigh❌ NoScheduled file ingestionStructured Streaming + foreachBatchSecondsMedium✅ YesComplex stateful pipelinesDelta Live Tables (DLT)ConfigurableHigh✅ Yes (expectations)Declarative, managed pipelinesMERGE only (batch)MinutesLow–Medium✅ YesSmall-to-medium upsert volumes DLT note: Delta Live Tables is Databricks' managed pipeline framework that handles the orchestration, monitoring, and retry logic described above declaratively. For teams starting fresh, DLT is worth evaluating before building the plumbing manually. Key Takeaways Medallion architecture separates concerns cleanly: Bronze for fidelity, Silver for correctness, Gold for consumption.Delta's transaction log is the foundation of all ACID guarantees — understanding it helps you debug merge conflicts, time travel, and VACUUM safely.Auto Loader is the right default for cloud file ingestion on Databricks — it handles exactly-once semantics and schema evolution automatically.MERGE with foreachBatch is the idiomatic pattern for deduplication and CDC in Spark Structured Streaming.Z-ORDER + Auto Optimize should be standard practice for Silver and Gold tables that receive frequent queries with selective filters.Deletion Vectors (Databricks 12.2+) make point deletes significantly cheaper — enable them for tables with GDPR or compliance requirements. References Delta Lake Documentation — Delta Lake Transaction Log — The official deep dive into how the _delta_log works internally.Databricks — Medallion ArchitectureDatabricks — Auto Loader (cloudFiles)Databricks — Delta Lake OPTIMIZE and Z-OrderingDatabricks — Auto Optimize (Optimized Writes + Auto Compaction)Databricks — Deletion VectorsDatabricks — Delta Live Tables OverviewStructured Streaming + foreachBatch — Apache Spark Docs"The Delta Lake Paper" — VLDB 2020 (Armbrust et al.)Databricks Blog — Diving Into Delta Lake: Unpacking the Transaction Log

By Jubin Soni, FBCS CORE