Kafka

This guide walks you through setting up a PySpark session configured for Kafka and UltiHash, enabling you to stream, process, and store data efficiently. You’ll learn how to configure environment variables, launch a local Kafka instance using Docker, produce and consume Kafka messages, and write structured data directly to an UltiHash bucket using PySpark. Ideal for real-time data pipelines or testing event-driven workflows.

Start PySpark session with config for Kafka

To begin, export your credentials and session variables, then launch a PySpark session with the necessary packages and configurations to connect with both Kafka and UltiHash’s S3-compatible storage.

(venv) ultihash@MacBookPro iceberg % export AWS_ACCESS_KEY_ID="TEST-USER"
export AWS_SECRET_ACCESS_KEY="SECRET"
export UH_CUSTOMER_ID="mem_cm6hrjwu703uf0sj12rcqfdpb"
export UH_ACCESS_TOKEN="vJZBfEHiX8cSSFws"
(venv) ultihash@MacBookPro iceberg % pyspark \\                           
  --packages org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.12.262,org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.1 \\
  --conf spark.sql.catalog.iceberg=org.apache.iceberg.spark.SparkCatalog \\
  --conf spark.sql.catalog.iceberg.type=hadoop \\
  --conf spark.sql.catalog.iceberg.warehouse=s3a://iceberg \\
  --conf spark.hadoop.fs.s3a.endpoint=http://127.0.0.1:8080 \\
  --conf spark.hadoop.fs.s3a.access.key=TEST-USER \\
  --conf spark.hadoop.fs.s3a.secret.key=SECRET \\
  --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \\
  --conf spark.hadoop.fs.s3a.path.style.access=true \\
  --conf spark.hadoop.fs.s3a.connection.ssl.enabled=false \\
  --conf spark.driver.bindAddress=127.0.0.1 \\
  --conf spark.driver.host=127.0.0.1

Write data to UltiHash through PySpark

Once your Spark session is running, you can create a simple DataFrame and write it as a Parquet file directly to a specified UltiHash bucket and path.

>>> from pyspark.sql import SparkSession
... 
... # Start your Spark session (should already be running with S3 configs from your CLI)
... spark = SparkSession.builder.appName("TestWriteToUltiHash").getOrCreate()
... 
... # Create a basic DataFrame
... df = spark.createDataFrame([
...     ("Alice", 25),
...     ("Bob", 32),
... ], ["name", "age"])
... 
... # Write the DataFrame as Parquet to UltiHash under the 'sample-data/' prefix in your 'kafka' bucket
... df.write.mode("overwrite").parquet("s3a://kafka/sample-data") #my bucket name is kafka
... 
... # Optional: Stop the Spark session (only if you're done)
... spark.stop()
... 
25/03/26 14:03:01 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
25/03/26 14:03:02 WARN MetricsConfig: Cannot locate configuration: tried hadoop-metrics2-s3a-file-system.properties,hadoop-metrics2.properties
>>>    

Create docker-compose.kafka.yalm

Set up a minimal Kafka and Zookeeper environment using Docker Compose. This configuration will run both services locally, exposing the standard ports for communication.

version: '3.8'

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

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

Start Kafka

With your Docker Compose file ready, bring up the Kafka and Zookeeper services in detached mode so they run in the background.

docker compose -f docker-compose.kafka.yml up -d

Create topic test-events

Inside the running Kafka container, create a topic called test-events that will be used to stream sample messages.

docker exec -it kafka kafka-topics --create \\
  --topic test-events \\
  --partitions 1 \\
  --replication-factor 1 \\
  --if-not-exists \\
  --bootstrap-server localhost:9092

Send message to Kafka

Use Kafka’s console producer to send test messages to the test-events topic. Enter each JSON message line by line in the interface.

docker exec -it kafka kafka-console-producer \\
  --broker-list localhost:9092 \\
  --topic test-events

#send in the interface line by line
{"user": "alice", "action": "click"}
{"user": "bob", "action": "scroll"}
{"user": "carol", "action": "purchase"}

Send the message from Kafka to UltiHash bucket through PySpark

Now, configure PySpark to consume the Kafka stream, parse the JSON messages, and write the structured data into UltiHash as Parquet files.

from pyspark.sql import SparkSession                             
... from pyspark.sql.functions import col, from_json
... from pyspark.sql.types import StringType, StructType
... 
... # Define schema of the Kafka message value
... schema = StructType().add("user", StringType()).add("action", StringType())
... 
... # Create a Spark session
... spark = SparkSession.builder.appName("KafkaToUltiHash").getOrCreate()
... 
... # Read streaming data from Kafka
... kafka_df = spark.readStream \\
...     .format("kafka") \\
...     .option("kafka.bootstrap.servers", "localhost:9092") \\
...     .option("subscribe", "test-events") \\
...     .option("startingOffsets", "earliest") \\
...     .load()
... 
... # Extract the 'value' from Kafka as a string and parse JSON
... parsed_df = kafka_df.selectExpr("CAST(value AS STRING) as json_str") \\
...     .select(from_json(col("json_str"), schema).alias("data")) \\
...     .select("data.*")
... 
... # Write to UltiHash
... query = parsed_df.writeStream \\
...     .format("parquet") \\
...     .option("path", "s3a://kafka/stream-output/") \\
...     .option("checkpointLocation", "/tmp/spark-checkpoint-kafka") \\
...     .outputMode("append") \\
...     .start()
... 
... query.awaitTermination()

Read what was written from PySpark

Finally, verify that the data was successfully written to UltiHash by reading it back as a DataFrame and displaying the results.

df = spark.read.parquet("s3a://kafka/stream-output/")
df.show()