Batch Processing vs Stream Processing — Key Differences Explained

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, count, date_trunc, round

spark = SparkSession.builder.appName("daily-revenue").getOrCreate()

# Read a day's worth of orders from S3 data lake
df = spark.read.parquet("s3://data-lake/orders/date=2026-03-25/")

# Step 1: Filter to completed orders only
completed = df.filter(col("status") == "completed")

# Step 2: Aggregate revenue by product category
revenue_by_category = (
    completed
    .groupBy("category", "region")
    .agg(
        sum("total").alias("revenue"),
        count("*").alias("order_count"),
        round(sum("total") / count("*"), 2).alias("avg_order_value")
    )
    .orderBy(col("revenue").desc())
)

# Step 3: Write to data warehouse (partitioned for efficient querying)
revenue_by_category.write     .mode("overwrite")     .partitionBy("region")     .parquet("s3://data-warehouse/daily-revenue/2026-03-25/")

print(f"Processed {completed.count():,} orders")
spark.stop()

-- dbt model: daily_revenue.sql
-- Runs nightly after raw order data lands in the warehouse

{{ config(
    materialized='incremental',
    partition_by={'field': 'order_date', 'data_type': 'date'},
    cluster_by=['region', 'category']
) }}

SELECT
    DATE(created_at)        AS order_date,
    region,
    category,
    COUNT(*)                AS order_count,
    SUM(total)              AS revenue,
    AVG(total)              AS avg_order_value,
    COUNT(DISTINCT user_id) AS unique_customers
FROM {{ source('raw', 'orders') }}
WHERE status = 'completed'
{% if is_incremental() %}
    -- Only process new data since last run
    AND DATE(created_at) > (SELECT MAX(order_date) FROM {{ this }})
{% endif %}
GROUP BY 1, 2, 3

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)  # Process in parallel across 4 workers
t_env = StreamTableEnvironment.create(env)

# Read from Kafka topic in real time
t_env.execute_sql("""
  CREATE TABLE orders (
    order_id STRING,
    user_id  STRING,
    total    DOUBLE,
    category STRING,
    ts       TIMESTAMP(3),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
  ) WITH (
    'connector' = 'kafka',
    'topic'     = 'order-events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id' = 'flink-revenue-processor',
    'scan.startup.mode' = 'latest-offset',
    'format'    = 'json'
  )
""")

# Real-time revenue windows: tumbling 1-minute windows
result = t_env.execute_sql("""
  SELECT
    TUMBLE_START(ts, INTERVAL '1' MINUTE) AS window_start,
    category,
    SUM(total)  AS revenue,
    COUNT(*)    AS order_count,
    AVG(total)  AS avg_order_value
  FROM orders
  GROUP BY
    TUMBLE(ts, INTERVAL '1' MINUTE),
    category
""")

result.print()

// Kafka Streams: real-time order counting per category
StreamsBuilder builder = new StreamsBuilder();

KStream<String, Order> orders = builder.stream("order-events");

KTable<Windowed<String>, Long> categoryCounts = orders
    .filter((key, order) -> "completed".equals(order.getStatus()))
    .groupBy((key, order) -> order.getCategory())
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count(Materialized.as("category-counts-store"));

// Write real-time counts back to Kafka
categoryCounts.toStream()
    .map((windowedKey, count) -> KeyValue.pair(
        windowedKey.key(),
        new CategoryCount(windowedKey.key(), count,
            windowedKey.window().startTime())
    ))
    .to("category-counts-output");

KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();