Batch-to-Streaming Bridge: One Lake for Both Pipelines

Overview

Your nightly ETL just finished loading yesterday’s numbers into the warehouse — and a customer on the fraud team’s dashboard is disputing a charge that happened forty seconds ago. The warehouse says the order doesn’t exist yet. It will, tomorrow at 2 a.m.

The batch-to-streaming bridge closes that gap. Batch pipelines chew through history in scheduled chunks; streaming pipelines chase events as they happen. The bridge makes both paths write compatible records into the same partitioned data lake, so a serving layer can answer “everything we know about this customer” without caring which path produced each row. It’s the Lambda architecture made concrete: a batch layer for completeness, a speed layer for freshness, merged at query time. The hard part is never the plumbing; it’s keeping schema, partitioning, and deduplication rules identical across two pipelines that evolve independently.

When to Use

• Analytics that need fresh events and deep history in a single query: fraud scoring, live dashboards over years of data.
• Migrating from batch ETL to streaming incrementally: run both in parallel and let the serving layer hide the transition.
• CDC-fed systems where the same entity arrives through a Kafka topic and through a nightly bulk load.
• Data lakes where batch backfills and streaming appends must coexist in the same tables.

When NOT to Use

• Pure streaming systems: run Kappa (below) or Kafka + Flink directly instead of maintaining a batch path you don’t need.
• Pure batch systems where hourly latency is acceptable: a speed layer is operational debt for zero benefit.
• Small teams that can’t afford two code paths. Lambda’s honest cost is that every schema change, every dedup rule, every partition decision gets implemented twice.
• Organizations that can adopt a lakehouse table format (Delta Lake, Apache Iceberg): ACID upserts give you most of this pattern with one pipeline instead of two.

Solution

Two paths, one lake. The streaming path lands events in S3 through Kafka Connect; the batch path lands bulk loads on the same partition scheme; Trino queries the union and deduplicates; Redis serves sub-second reads for the freshest state.

Kafka Connect S3 sink for the streaming path

{
  "name": "s3-customers-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "4",
    "topics": "customer-events",
    "s3.bucket.name": "data-lake-shop",
    "s3.region": "us-east-1",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "parquet.codec": "snappy",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "partition.duration.ms": "3600000",
    "path.format": "year=YYYY/month=MM/day=dd/hour=HH",
    "locale": "en_US",
    "timezone": "UTC",
    "flush.size": "10000",
    "rotate.interval.ms": "600000",
    "schema.compatibility": "BACKWARD",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.enhanced.avro.schema.support": "true"
  }
}

The two fields that matter for the bridge are path.format, which decides the partition layout every other path must match, and schema.compatibility, which the shared schema registry enforces on both paths.

Batch path writing to the same data lake

# batch_to_lake.py — batch ETL writing to the same S3 path as streaming
import boto3
import pandas as pd
from datetime import datetime

class BatchToLake:
    def __init__(self, bucket, prefix):
        self.s3 = boto3.client('s3')
        self.bucket = bucket
        self.prefix = prefix

    def load_batch(self, df, table_name, batch_hour):
        """Write batch data to the same S3 path format as streaming."""
        # Match the streaming partition format exactly, hour included:
        # year=YYYY/month=MM/day=dd/hour=HH
        parts = batch_hour.strftime('%Y/%m/%d/%H').split('/')
        year, month, day, hour = parts

        s3_key = (
            f"{self.prefix}/{table_name}"
            f"/year={year}/month={month}/day={day}/hour={hour}"
            f"/batch_{batch_hour.strftime('%Y%m%d%H')}.parquet"
        )

        parquet_buffer = df.to_parquet(engine='pyarrow', compression='snappy')
        self.s3.put_object(Bucket=self.bucket, Key=s3_key, Body=parquet_buffer)
        return s3_key

    def list_partitions(self, table_name, date):
        """List files for a given date — both batch and streaming land here."""
        year, month, day = date.strftime('%Y'), date.strftime('%m'), date.strftime('%d')
        prefix = f"{self.prefix}/{table_name}/year={year}/month={month}/day={day}/"

        response = self.s3.list_objects_v2(Bucket=self.bucket, Prefix=prefix)
        return [
            {
                'key': obj['Key'],
                'size': obj['Size'],
                'source': 'batch' if 'batch_' in obj['Key'] else 'streaming'
            }
            for obj in response.get('Contents', [])
        ]

The batch_ filename prefix is deliberate: it lets the serving layer tag _source without inspecting file metadata, and it makes debugging partition contents trivial. The hour segment matters more than it looks; drop it and the batch files land outside the table’s declared partitions, invisible to Trino.

Unified serving layer with Trino

-- serving_layer.sql — query batch and streaming data together
-- Both paths write to the same partitioned table in S3;
-- Trino treats them as one table.

CREATE EXTERNAL TABLE IF NOT EXISTS lake.customers (
    id BIGINT,
    email VARCHAR,
    status VARCHAR,
    is_active BOOLEAN,
    created_at TIMESTAMP,
    _source VARCHAR,
    _ingested_at TIMESTAMP
)
STORED AS PARQUET
LOCATION 's3://data-lake-shop/customers/'
PARTITIONED BY (
    year VARCHAR,
    month VARCHAR,
    day VARCHAR,
    hour VARCHAR
);

-- Discover partitions written since the last repair
MSCK REPAIR TABLE lake.customers;

-- One query over both paths
SELECT
    _source,
    COUNT(*) AS row_count,
    MIN(_ingested_at) AS earliest,
    MAX(_ingested_at) AS latest
FROM lake.customers
WHERE year = '2026' AND month = '08' AND day = '19'
GROUP BY _source;

-- Merge batch and streaming into a deduplicated view:
-- prefer streaming (fresher), fall back to batch
CREATE OR REPLACE VIEW warehouse.customers_unified AS
SELECT
    id,
    email,
    status,
    is_active,
    created_at,
    ROW_NUMBER() OVER (
        PARTITION BY id
        ORDER BY
            CASE _source WHEN 'streaming' THEN 1 WHEN 'batch' THEN 2 END,
            _ingested_at DESC
    ) AS rn
FROM lake.customers
WHERE year = CAST(YEAR(CURRENT_DATE) AS VARCHAR)
  AND month = LPAD(CAST(MONTH(CURRENT_DATE) AS VARCHAR), 2, '0');

CREATE OR REPLACE VIEW warehouse.customers_latest AS
SELECT id, email, status, is_active, created_at
FROM warehouse.customers_unified
WHERE rn = 1;

The dedup rule encodes the pattern’s core assumption: streaming rows are fresher, batch rows are the safety net. CASE _source ranks streaming first, and _ingested_at DESC breaks ties inside each source.

Flink processor for the speed layer

// StreamingProcessor.java — Flink processor for the speed layer
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.formats.avro.AvroDeserializationSchema;
import org.apache.flink.formats.avro.AvroSerializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import java.time.Duration;

public class StreamingProcessor {
    public static void main(String[] args) throws Exception {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();

        KafkaSource<Customer> source = KafkaSource.<Customer>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("customer-events")
            .setGroupId("flink-processor")
            .setValueOnlyDeserializer(
                AvroDeserializationSchema.forSpecific(Customer.class)
            )
            .build();

        DataStream<Customer> events = env.fromSource(
            source,
            // Bounded out-of-orderness: events more than 5 min late
            // go to the side output, not the main stream
            WatermarkStrategy.<Customer>forBoundedOutOfOrderness(Duration.ofMinutes(5)),
            "customer-events"
        );

        DataStream<CustomerEnriched> enriched = events
            .map(customer -> {
                var e = new CustomerEnriched();
                e.setId(customer.getId());
                e.setEmail(customer.getEmail().toLowerCase());
                e.setStatus(customer.getStatus());
                e.setActive("active".equals(customer.getStatus()));
                e.setProcessedAt(System.currentTimeMillis());
                return e;
            })
            .name("enrich");

        // Real-time consumers downstream of Kafka
        KafkaSink<CustomerEnriched> kafkaSink = KafkaSink.<CustomerEnriched>builder()
            .setBootstrapServers("kafka:9092")
            .setRecordSerializer(
                KafkaRecordSerializationSchema.builder()
                    .setTopic("customer-events-enriched")
                    .setValueSerializationSchema(
                        AvroSerializationSchema.forSpecific(CustomerEnriched.class)
                    )
                    .build()
            )
            .build();

        enriched.sinkTo(kafkaSink).name("kafka-sink");

        // Speed layer: latest state per key for sub-second reads
        enriched.addSink(new RedisSink<>()).name("redis-sink");

        env.execute("customer-stream-processor");
    }
}

Note the watermarks: forBoundedOutOfOrderness(5 minutes) is the streaming half of the late-data policy. Events that arrive later than the watermark get routed to a side output (or dropped); the batch path picks them up on the next run. That’s the bridge working as designed: late data doesn’t disappear, it just arrives through the slower door.

Python unified consumer

# unified_consumer.py — history from Trino, fresh state from the speed layer
import json
import trino
import pandas as pd
import redis

class UnifiedConsumer:
    def __init__(self, trino_host, redis_client):
        self.trino = trino.dbapi.connect(host=trino_host, port=443, user="analyst")
        self.redis = redis_client  # the speed layer Flink writes to

    def get_historical(self, customer_id):
        """Get deduplicated history from the batch layer via Trino."""
        query = "SELECT * FROM warehouse.customers_latest WHERE id = ?"
        return pd.read_sql(query, self.trino, params=[customer_id])

    def get_realtime(self, customer_id):
        """Get the latest state from the speed layer (keyed lookup, not a scan)."""
        raw = self.redis.get(f"customer:{customer_id}")
        return json.loads(raw) if raw else None

    def get_unified_view(self, customer_id):
        """Combine batch history with the freshest streaming state."""
        historical = self.get_historical(customer_id)
        result = historical.iloc[0].to_dict() if not historical.empty else {}

        latest = self.get_realtime(customer_id)
        if latest:
            # Streaming overrides only the fields it actually carries
            for field in ('email', 'status', 'is_active'):
                if field in latest:
                    result[field] = latest[field]
            result['_source'] = 'merged'

        return result or None

Two things changed versus a naive implementation. The speed-layer read is a keyed GET, not a Kafka consumer; re-consuming the topic to find one customer’s latest event would scan other customers’ traffic and block. And the merge applies streaming fields selectively, so a partial update can’t wipe fields it doesn’t carry — a status-only event never touches email.

Schema alignment between batch and streaming

# schema_alignment.py — one model both paths must produce
from pydantic import BaseModel
from datetime import datetime
from typing import Literal

class CustomerRecord(BaseModel):
    """Shared schema for both batch and streaming paths."""
    id: int
    email: str
    status: Literal["active", "inactive", "banned", "suspended"]
    is_active: bool
    created_at: datetime
    _source: Literal["batch", "streaming"]
    _ingested_at: datetime

class BatchProducer:
    def produce(self, row):
        return CustomerRecord(
            id=row['id'],
            email=row['email'].lower().strip(),
            status=row['status'],
            is_active=row['status'] == 'active',
            created_at=row['created_at'],
            _source="batch",
            _ingested_at=datetime.now()
        )

class StreamingProducer:
    def produce(self, event):
        # CDC envelope: the row lives under 'after'
        after = event['after']
        return CustomerRecord(
            id=after['id'],
            email=after['email'].lower().strip(),
            status=after['status'],
            is_active=after['status'] == 'active',
            created_at=after['created_at'],
            _source="streaming",
            _ingested_at=datetime.now()
        )

Both producers run the same normalization: lowercase email, derived is_active, tagged _source. If one path lowercases and the other doesn’t, dedup by id still works but every downstream filter on email silently splits. In production the schema registry enforces this at the wire; the shared model enforces it in code.

Variants

Kappa architecture (streaming-only)

# kappa_architecture.py — replace batch with replayed streaming
from kafka import KafkaConsumer
from datetime import datetime

class KappaArchitecture:
    """No batch layer. All data flows through streaming; historical
    reprocessing replays the Kafka topic from the beginning."""
    def reprocess(self, kafka_topic, new_processor):
        # Fresh consumer group starts at the earliest offset
        consumer = KafkaConsumer(
            kafka_topic,
            auto_offset_reset='earliest',
            enable_auto_commit=False,
            group_id=f'reprocess-{datetime.now().timestamp()}'
        )

        for message in consumer:
            new_processor.process(message.value)

Kappa removes the second code path entirely, but it only works if retention covers your history or you compact + replay faithfully, and if “reprocessing” doesn’t need a different compute model than live processing. Teams pick Lambda not because it’s elegant but because their batch jobs genuinely do something streaming can’t: joins against snapshots, model training, regulatory reconciliation.

Lakehouse table format (the modern answer)

Delta Lake and Apache Iceberg deserve a mention because they collapse much of this pattern into the storage layer: ACID upserts mean streaming and batch writers can update the same table without partition conventions, _source tags, or query-time dedup; the engine merges by primary key. If you’re starting fresh on a lakehouse, you may not need the bridge at all. The pattern stays relevant on plain Parquet lakes, on existing Hive-style tables, and anywhere migration cost rules out a format switch.

Hybrid with materialized views

-- materialized_view.sql — precompute the merge for hot queries
CREATE MATERIALIZED VIEW warehouse.customer_stats_hourly AS
SELECT
    DATE_TRUNC('hour', _ingested_at) AS hour,
    _source,
    COUNT(*) AS event_count,
    COUNT(DISTINCT id) AS unique_customers,
    SUM(CASE WHEN is_active THEN 1 ELSE 0 END) AS active_count
FROM lake.customers
WHERE year = CAST(YEAR(CURRENT_DATE) AS VARCHAR)
  AND month = LPAD(CAST(MONTH(CURRENT_DATE) AS VARCHAR), 2, '0')
GROUP BY DATE_TRUNC('hour', _ingested_at), _source;

REFRESH MATERIALIZED VIEW warehouse.customer_stats_hourly;

Best Practices

• Use a shared schema registry — both paths must produce the same Avro/Protobuf schema, enforced at write time. See the schema evolution pattern for compatibility rules.
• Tag records with _source — distinguish batch from streaming in every query; it’s also your first debugging tool when counts don’t reconcile.
• Match partition formats exactly — both paths write year=YYYY/month=MM/day=dd/hour=HH. One missing segment and the serving layer reads half the data.
• Make both paths idempotent — re-runs and backfills are routine; dedup keys (id + _ingested_at) must survive them.
• Monitor lag on both sides — streaming lag means the speed layer serves stale data; batch lag means history gaps. Alert on both.
• Prefer Kappa or a lakehouse format when starting fresh — maintaining two pipelines is the real cost of Lambda.

Common Mistakes

• Schema drift between paths: batch lowercases email, streaming doesn’t; filters split silently. Enforce one shared model.
• Partition mismatch: batch writes day= granularity while the table expects hour=; batch files become invisible to the query engine. Align formats down to the last segment.
• No dedup at the serving layer: a customer appears twice, once per path. ROW_NUMBER() over id ordered by source freshness fixes it.
• Reading fresh state by re-consuming Kafka: scanning the topic for one key blocks and swallows other keys’ events. The speed layer exists for keyed lookups — use it.
• Streaming-only with no batch fallback: when the stream pipeline fails (and it will), there’s nothing to reconcile against. Keep the batch path even if it’s slow.

Further Reading

• Confluent: S3 Sink Connector — partitioning, formats, and schema registry integration for the streaming write path.
• AWS Prescriptive Guidance: Build a Lambda architecture — the managed-service version of this pattern.
• Trino documentation: Hive connector — external tables, partitions, and MSCK REPAIR semantics.
• Apache Iceberg — the table format that absorbs much of this pattern.
• Companion examples — runnable schema alignment and serving-layer dedup in Python and JavaScript with tests.