Build batch pipelines with Spark and stream with Kafka, Flink

At large enough scale you will need Spark, Airflow, Kafka and Flink. This lesson walks through building both kinds of pipeline.

The code here is a sample to read and adapt. It has not been run on a real cluster, so try it on your own machine first.

What are the steps for a batch pipeline?#

1. Settle the requirements. Who uses the result, how fresh it must be, where the data lives.
2. Load data into the lake. Copy it to object storage in a columnar format such as Parquet, in folders by date.
3. Write the transform job. Each run handles a clear period, such as one day.
4. Make the job idempotent. Re-running a day must give the same result. Overwrite that day's partition.
5. Orchestrate with Airflow. Declare the schedule, order, retries and failure alerts.
6. Watch quality. Measure run time and rows in and out, check for nulls, duplicates and wrong totals. Have a way to backfill ready.

# daily_revenue.py: revenue per shop for one day
import sys
from pyspark.sql import SparkSession, functions as F

day = sys.argv[1]  # for example 2026-10-25
spark = (SparkSession.builder.appName("daily_revenue")
         .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
         .getOrCreate())

orders = spark.read.parquet(f"s3a://lake/orders/dt={day}/")

daily = (orders
    .where(F.col("status") == "PAID")
    .groupBy("shop_id")
    .agg(F.count("*").alias("so_don"),
         F.sum("amount").alias("doanh_thu")))

# Idempotent: overwrite only this day's partition
(daily.withColumn("dt", F.lit(day))
      .write.mode("overwrite")
      .partitionBy("dt")
      .parquet("s3a://warehouse/daily_revenue/"))

Without partitionOverwriteMode=dynamic, the overwrite wipes every other partition.

# dags/daily_revenue.py: Airflow 3
from datetime import datetime
from airflow.sdk import dag, task

# Runs at 02:00 each night and fills in missed days (catchup)
@dag(schedule="0 2 * * *", start_date=datetime(2026, 10, 1), catchup=True)
def daily_revenue():

    @task.bash(retries=2)
    def run_spark():
        # {{ ds }} is the run's logical date, so a backfill hits the right day
        return "spark-submit daily_revenue.py {{ ds }}"

    run_spark()

daily_revenue()

What are the steps for a stream pipeline?#

1. Define the event. An id, the event time and the data. Pick a key to partition by, such as shop_id.
2. Put events on Kafka. Choose the partition count by the throughput you need, since partitions set the parallelism.
3. Choose the engine. Flink for large state and exact event time. Kafka Streams for small logic inside a Java service. Spark if the team knows Spark.
4. Declare time and windows. Set event time, the watermark and the window type.
5. Turn on checkpoints and pick a destination. Write with upserts by key so reprocessing does not double results.
6. Operate it. Watch consumer lag, checkpoint time and backpressure. Try stopping a worker to see how it recovers.

# producer.py: send orders to the "orders" topic
import json, time
from confluent_kafka import Producer

p = Producer({"bootstrap.servers": "localhost:9092"})

def send_order(order_id, shop_id, amount):
    event = {"order_id": order_id, "shop_id": shop_id,
             "amount": amount,
             "ts": int(time.time() * 1000)}  # event time, milliseconds
    # key = shop_id: one shop's orders go to one partition
    p.produce("orders", key=shop_id, value=json.dumps(event))

send_order("A1021", "shop-7", 249000)
p.flush()

-- 1) Declare the source, allowing events up to 5 seconds late
CREATE TABLE orders (
  order_id STRING,
  shop_id  STRING,
  amount   BIGINT,
  ts       BIGINT,
  ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
  WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'localhost:9092',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json'
);

-- 2) Count orders and revenue per shop in 1-minute windows
SELECT window_start, window_end, shop_id,
       COUNT(*) AS so_don, SUM(amount) AS doanh_thu
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(ts_ltz), INTERVAL '1' MINUTE))
GROUP BY window_start, window_end, shop_id;

Epoch milliseconds do not parse directly as TIMESTAMP_LTZ, so the sample keeps ts BIGINT and converts it.

What should you learn next, and in what order?#

Two exercises to practise:

• Batch exercise. Use PySpark on your own machine to compute daily revenue from a CSV of orders.
• Stream exercise. Run Kafka and Flink with Docker Compose, write a producer that simulates orders and count orders per minute with Flink SQL.

Common mistakes#

• Appending instead of overwriting the partition. Every re-run doubles the numbers.
• Forgetting catchup and the logical date. A backfill computes the wrong day if the job uses the current time instead of {{ ds }}.
• A skewed key. One huge shop lands in one partition and chokes the whole system.
• Never stopping a machine on purpose. Only by stopping a worker do you learn whether checkpoints recover correctly.