Microbatching from Kafka to Hudi dataset

In this article, we will try to learn how can we implement microbatch using Kafka, Apache Spark and Apache Hudi.

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StringType, LongType, TimestampType

spark = SparkSession.builder \
    .appName("KafkaToHudi") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog") \
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension") \
    .getOrCreate()

spark.sparkContext.setLogLevel("WARN")

schema = StructType() \
    .add("ride_id", StringType()) \
    .add("driver_id", StringType()) \
    .add("fare", LongType()) \
    .add("city", StringType()) \
    .add("ts", StringType())

kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "rides") \
    .option("startingOffsets", "earliest") \
    .load()

parsed_df = kafka_df.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

def write_to_hudi(batch_df, batch_id):
    print(f"--- Processing batch {batch_id}, count: {batch_df.count()} ---")
    batch_df.show()

    if batch_df.count() == 0:
        print(f"Batch {batch_id} is empty, skipping.")
        return

    hudi_options = {
        "hoodie.table.name": "rides",
        "hoodie.datasource.write.recordkey.field": "ride_id",
        "hoodie.datasource.write.partitionpath.field": "city",
        "hoodie.datasource.write.precombine.field": "ts",
        "hoodie.datasource.write.operation": "upsert",
        "hoodie.datasource.hive_sync.enable": "false",
    }

    batch_df.write \
        .format("hudi") \
        .options(**hudi_options) \
        .mode("append") \
        .save("/opt/data/rides_hudi")

query = parsed_df.writeStream \
    .foreachBatch(write_to_hudi) \
    .option("checkpointLocation", "/opt/data/checkpoints/rides") \
    .trigger(processingTime="15 seconds") \
    .start()

print("Streaming job started. Waiting for data...")
query.awaitTermination()

Here in this script the thing to note is that the partition is by cityId and in this line “parsed_df.writeStream” we are setting up microbatching at 15secs.

So it’s pretty straightforward now, where we don’t really need to manage this microbatching. Spark does all the hardwork for us-

Here are the other files used in this program-

docker-compose.yaml

version: '3.8'

services:

  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

  spark:
    image: spark-hudi:latest
    container_name: spark
    depends_on:
      - kafka
    ports:
      - "4040:4040"
    volumes:
      - ./jobs:/opt/jobs
      - ./data:/opt/data
    command: tail -f /dev/null
    environment:
      - SPARK_MODE=master

Dockerfile(Spark job)

FROM apache/spark:3.4.1

USER root

# Download Hudi + Kafka connector jars
RUN cd /opt/spark/jars && \
    wget -q https://repo1.maven.org/maven2/org/apache/hudi/hudi-spark3.4-bundle_2.12/0.14.0/hudi-spark3.4-bundle_2.12-0.14.0.jar && \
    wget -q https://repo1.maven.org/maven2/org/apache/spark/spark-sql-kafka-0-10_2.12/3.4.1/spark-sql-kafka-0-10_2.12-3.4.1.jar && \
    wget -q https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.4.0/kafka-clients-3.4.0.jar && \
    wget -q https://repo1.maven.org/maven2/org/apache/spark/spark-token-provider-kafka-0-10_2.12/3.4.1/spark-token-provider-kafka-0-10_2.12-3.4.1.jar && \
    wget -q https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/2.11.1/commons-pool2-2.11.1.jar

USER spark

Now, we dispacthed these 3messages through Kafka topic

{"ride_id":"r001","driver_id":"d01","fare":120,"city":"kolkata","ts":"2026-06-14 10:00:00"}
{"ride_id":"r002","driver_id":"d02","fare":85,"city":"mumbai","ts":"2026-06-14 10:01:00"}
{"ride_id":"r003","driver_id":"d01","fare":200,"city":"kolkata","ts":"2026-06-14 10:02:00"}

As soon as our Spark job spawned in a single micro-batch all three rides were picked-

Spark Job processing micro-batch

Now, I sent two more messages one by one with some time gap

{"ride_id":"r004","driver_id":"d03","fare":310,"city":"delhi","ts":"2026-06-14 10:03:00"}
{"ride_id":"r001","driver_id":"d01","fare":999,"city":"kolkata","ts":"2026-06-14 10:04:00"}

Multi-batch working

Now, if you see here there are two batches this time each processing a different data-

Finally, we have all partitions created in our filesystem-