Skip to content

Kafka

For engineers who feed a Kafka topic. It teaches an event stream as JSON lines, and the producer block that sends events to a broker. Each stream_dict call continues the seeded sequence, so a second call yields different payloads under the same event_id range.

Prerequisite and run

Install rand-engine, then scaffold and run the file. Its default path needs neither a broker nor kafka-python:

Get the file:

pip install rand-engine
rand-engine get-started kafka
python kafka-example.py

It writes kafka-example.py to the current directory. --out DIR writes it to DIR instead, created when missing. An existing file is never overwritten. Running it as-is prints 100 JSON events, one line per event, to stdout.

# Databricks notebook source
# Kafka: an event stream as JSON lines on stdout. Runs as-is with plain python; no broker or client needed.

# COMMAND ----------
# Describe the events and the switches. The producer stays off until you turn it on.

import json
from itertools import islice

from rand_engine import DataGenerator

SEND_TO_KAFKA = False  # pip install kafka-python, then set True
BOOTSTRAP_SERVERS = "localhost:9092"
EVENT_SPEC = {
  "event_id": {"method": "pk", "kwargs": {"start": 1}},
  "kind": {"method": "distincts", "kwargs": {"distincts": ["click", "view", "buy"]}},
  "user": {"method": "int_zfilled", "kwargs": {"length": 6}},
  "amount": {"method": "floats", "kwargs": {"min": 1, "max": 300, "decimals": 2}},
  "ts": {"method": "unix_timestamps", "kwargs": {"start": "2024-01-01", "end": "2024-12-31"}},
}

# COMMAND ----------
# Print 100 events, one JSON line each. stream_dict adds a wall-clock timestamp_created that
# would change on every run, so it is dropped; ts is the seeded event time.

events = DataGenerator(EVENT_SPEC, seed=7).size(100)
for event in islice(events.stream_dict(min_throughput=500, max_throughput=1000), 100):
  event.pop("timestamp_created")
  print(json.dumps(event))

# COMMAND ----------
# Producer, off by default. Set SEND_TO_KAFKA = True and BOOTSTRAP_SERVERS first.

if SEND_TO_KAFKA:
  from kafka import KafkaProducer

  producer = KafkaProducer(bootstrap_servers=BOOTSTRAP_SERVERS)
  for event in islice(events.stream_dict(min_throughput=500, max_throughput=1000), 100):
    event.pop("timestamp_created")
    producer.send("events", json.dumps(event).encode())
  producer.flush()

Enable the producer

The producer is off by default. To send the events to a topic named events:

  1. Install the client: pip install kafka-python.
  2. In the file, set SEND_TO_KAFKA = True and set BOOTSTRAP_SERVERS to your broker.
  3. Run the file again.

What it shows

  • Describe: the event spec, and the two switches, SEND_TO_KAFKA and BOOTSTRAP_SERVERS.
  • Print: 100 events from stream_dict, one JSON line each. The wall-clock timestamp_created is dropped, so the seeded ts stays the event time.
  • Producer: the block that runs only when SEND_TO_KAFKA is set.

Next steps