JSONL for Data Streaming

Kafka Producer (Python)

Produce JSON events to Kafka topics for real-time processing. Use confluent-kafka-python, the officially supported client wrapping the C library librdkafka - it is the maintained option and delivers far higher throughput than a pure-Python client.

from confluent_kafka import Producer
import json
import uuid
from datetime import datetime, timezone

# Create producer (librdkafka handles batching, retries and compression)
producer = Producer({
    'bootstrap.servers': 'localhost:9092',
    'acks': 'all',                  # Wait for all in-sync replicas
    'compression.type': 'gzip',     # Compress messages
    'enable.idempotence': True,     # Exactly-once semantics per partition
    'linger.ms': 20,                # Small batching window
    'retries': 3
})

def delivery_report(err, msg):
    """Called once per message, on success or permanent failure."""
    if err is not None:
        print(f"Failed to send event: {err}")
    else:
        print(f"Sent to partition {msg.partition()} at offset {msg.offset()}")

# Send user events
def send_user_event(user_id, action, metadata=None):
    event = {
        'event_id': str(uuid.uuid4()),
        'user_id': user_id,
        'action': action,
        'timestamp': datetime.now(timezone.utc).isoformat(),
        'metadata': metadata or {}
    }

    # Send to topic with user_id as key for partitioning
    producer.produce(
        'user-events',
        key=str(user_id),
        value=json.dumps(event).encode('utf-8'),
        on_delivery=delivery_report
    )

    # Serve delivery callbacks without blocking the producer
    producer.poll(0)

# Usage
send_user_event(12345, 'login', {'ip': '192.168.1.100', 'device': 'mobile'})
send_user_event(12345, 'view_product', {'product_id': 'prod_789', 'category': 'electronics'})
send_user_event(12345, 'add_to_cart', {'product_id': 'prod_789', 'quantity': 1})

# Block until every queued message has been delivered
producer.flush(10)

Kafka Consumer (Python)

The same library handles the consume side. Poll in a loop, decode each message as one JSON object, and commit offsets only after the event is safely processed.

from confluent_kafka import Consumer, KafkaError
import json

# Create consumer
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'event-processor-group',
    'auto.offset.reset': 'earliest',  # Start from beginning if no offset
    'enable.auto.commit': False,      # Commit only after successful processing
    'max.poll.interval.ms': 300000
})

consumer.subscribe(['user-events'])

def process_purchase(event):
    # Send confirmation email
    # Update inventory
    # Record in database
    pass

def update_cart(event):
    # Update user's shopping cart
    pass

def track_analytics(event):
    # Send to analytics platform
    pass

# Process events in real-time
try:
    while True:
        message = consumer.poll(timeout=1.0)
        if message is None:
            continue
        if message.error():
            if message.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise Exception(message.error())

        event = json.loads(message.value().decode('utf-8'))

        print(f"Processing event: {event['action']} for user {event['user_id']}")
        print(f"Partition: {message.partition()}, Offset: {message.offset()}")

        # Process based on action type
        if event['action'] == 'purchase':
            process_purchase(event)
        elif event['action'] == 'add_to_cart':
            update_cart(event)
        elif event['action'] == 'view_product':
            track_analytics(event)

        # At-least-once delivery: commit after the work is done
        consumer.commit(message=message, asynchronous=False)
finally:
    consumer.close()

Legacy reference only: kafka-python (do not use for new work)

Shown purely so you can recognise it in older code and migrate away from it. The consumer group id, topic and offset-reset settings map one-for-one onto the confluent-kafka configuration above.

# DEPRECATED - kafka-python is effectively unmaintained.
# Migrate this to confluent_kafka.Consumer (see above).
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'user-events',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    group_id='event-processor-group',
    auto_offset_reset='earliest'
)

for message in consumer:
    event = message.value
    print(f"Processing event: {event['action']} for user {event['user_id']}")

Kafka Streams (Java)

Stream processing with stateful transformations and aggregations.

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

public class UserEventProcessor {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        ObjectMapper mapper = new ObjectMapper();

        // Read events stream
        KStream<String, String> events = builder.stream("user-events");

        // Parse JSON and filter
        KStream<String, JsonNode> parsedEvents = events
            .mapValues(value -> {
                try {
                    return mapper.readTree(value);
                } catch (Exception e) {
                    return null;
                }
            })
            .filter((key, value) -> value != null);

        // Count purchases by user (windowed aggregation)
        KTable<Windowed<String>, Long> purchaseCounts = parsedEvents
            .filter((key, value) -> value.get("action").asText().equals("purchase"))
            .groupBy((key, value) -> value.get("user_id").asText())
            .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
            .count();

        // Alert on high-value purchases
        parsedEvents
            .filter((key, value) ->
                value.get("action").asText().equals("purchase") &&
                value.get("amount").asDouble() > 1000.0
            )
            .to("high-value-purchases");

        // Calculate average cart value
        parsedEvents
            .filter((key, value) -> value.get("action").asText().equals("add_to_cart"))
            .groupByKey()
            .aggregate(
                () -> new CartAggregate(),
                (key, value, aggregate) -> aggregate.add(value),
                Materialized.as("cart-aggregates")
            );

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

KafkaJS (Node.js)

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

// Producer
const producer = kafka.producer();
await producer.connect();

async function sendEvent(topic, key, event) {
  await producer.send({
    topic,
    messages: [
      {
        key,
        value: JSON.stringify(event),
        headers: {
          'correlation-id': event.correlationId,
        },
      },
    ],
  });
}

// Consumer
const consumer = kafka.consumer({ groupId: 'event-processors' });
await consumer.connect();
await consumer.subscribe({ topic: 'user-events', fromBeginning: false });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value.toString());

    console.log({
      partition,
      offset: message.offset,
      value: event,
    });

    // Process event
    await processEvent(event);
  },
});

async function processEvent(event) {
  switch (event.action) {
    case 'purchase':
      await handlePurchase(event);
      break;
    case 'refund':
      await handleRefund(event);
      break;
    default:
      console.log(`Unknown action: ${event.action}`);
  }
}

Kafka Connect JSONL Sink

Export Kafka topics to JSONL files for archival or batch processing.

{
  "name": "jsonl-file-sink",
  "config": {
    "connector.class": "FileStreamSink",
    "tasks.max": "1",
    "file": "/data/events.jsonl",
    "topics": "user-events",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

# Load connector
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @jsonl-sink-connector.json

# Result: Continuous JSONL export
# /data/events.jsonl contains:
{"event_id": "evt_001", "user_id": 123, "action": "login", "timestamp": "2026-01-15T10:00:00Z"}
{"event_id": "evt_002", "user_id": 124, "action": "view", "timestamp": "2026-01-15T10:00:05Z"}
{"event_id": "evt_003", "user_id": 123, "action": "purchase", "timestamp": "2026-01-15T10:01:00Z"}

Debezium CDC to Kafka

Stream database changes as JSONL events for real-time data synchronization.

# Debezium PostgreSQL Connector Configuration
{
  "name": "postgres-cdc-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres.example.com",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "secret",
    "database.dbname": "myapp",
    "database.server.name": "myapp-db",
    "table.include.list": "public.users,public.orders",
    "plugin.name": "pgoutput",

    # JSON output format
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",

    # Kafka topic configuration
    "topic.prefix": "cdc",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false"
  }
}

# Example CDC event format
{
  "before": null,
  "after": {
    "id": 12345,
    "email": "[email protected]",
    "name": "John Doe",
    "created_at": "2026-01-15T10:00:00Z"
  },
  "source": {
    "version": "1.9.0",
    "connector": "postgresql",
    "name": "myapp-db",
    "ts_ms": 1705315200000,
    "db": "myapp",
    "schema": "public",
    "table": "users"
  },
  "op": "c",  // c=create, u=update, d=delete
  "ts_ms": 1705315200000
}

CDC Consumer

Read the change topic with confluent-kafka-python and branch on the Debezium operation code.

from confluent_kafka import Consumer
import json

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'cdc-processor',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False
})

consumer.subscribe(['cdc.public.users'])

while True:
    message = consumer.poll(timeout=1.0)
    if message is None or message.error():
        continue

    # Debezium tombstones (deletes) carry a null value - skip them
    if message.value() is None:
        continue

    event = json.loads(message.value().decode('utf-8'))

    operation = event['op']
    table = event['source']['table']

    if operation == 'c':  # Create
        handle_insert(event['after'])
    elif operation == 'u':  # Update
        handle_update(event['before'], event['after'])
    elif operation == 'd':  # Delete
        handle_delete(event['before'])

    consumer.commit(message=message, asynchronous=False)

def handle_insert(record):
    """Sync new record to search index, cache, etc."""
    print(f"New record: {record}")
    # Update Elasticsearch
    # Invalidate cache
    # Send to analytics

def handle_update(before, after):
    """Handle record updates"""
    print(f"Updated: {before} -> {after}")
    # Track field changes
    # Update downstream systems

def handle_delete(record):
    """Handle record deletions"""
    print(f"Deleted: {record}")
    # Remove from search index
    # Archive for compliance

CDC Use Cases

WebSocket JSONL Streaming

// Server (Node.js with ws)
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', (ws) => {
  console.log('Client connected');

  // Stream events to client as JSONL
  const interval = setInterval(() => {
    const event = {
      type: 'stock_update',
      symbol: 'AAPL',
      price: (Math.random() * 200).toFixed(2),
      timestamp: new Date().toISOString()
    };

    // Send as JSON string (one per message)
    ws.send(JSON.stringify(event));
  }, 1000);

  ws.on('close', () => {
    clearInterval(interval);
    console.log('Client disconnected');
  });
});

// Client (Browser)
const ws = new WebSocket('ws://localhost:8080');

ws.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Received:', data);

  // Update UI with real-time data
  updateStockPrice(data.symbol, data.price);
};

ws.onopen = () => {
  console.log('Connected to stream');
};

ws.onerror = (error) => {
  console.error('WebSocket error:', error);
};

Server-Sent Events (SSE)

// Server (Express)
const express = require('express');
const app = express();

app.get('/events', (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  // Send events every second
  const interval = setInterval(() => {
    const event = {
      id: Date.now(),
      type: 'update',
      data: {
        metric: 'active_users',
        value: Math.floor(Math.random() * 1000)
      }
    };

    // SSE format: data: JSON\n\n
    res.write(`data: ${JSON.stringify(event)}\n\n`);
  }, 1000);

  req.on('close', () => {
    clearInterval(interval);
  });
});

app.listen(3000);

// Client (Browser)
const eventSource = new EventSource('/events');

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Event received:', data);

  // Update dashboard
  updateMetric(data.data.metric, data.data.value);
};

eventSource.onerror = (error) => {
  console.error('SSE error:', error);
  eventSource.close();
};

Redpanda (Kafka API Compatible)

Redpanda speaks the Kafka wire protocol, so every client on this page keeps working unchanged - including the confluent-kafka producer above. The difference is operational: one C++ binary, no JVM and no ZooKeeper. Its CLI, rpk, pipes JSONL files in and out of a topic directly, which makes it a genuinely pleasant local test harness.

# Create a topic with 6 partitions
rpk topic create user-events -p 6

# Produce JSONL: one JSON object per line, straight from a file
rpk topic produce user-events < events.jsonl

# Consume it back as JSONL (values only, no envelope)
rpk topic consume user-events --format '%v\n' --num 3
{"event_id": "evt_001", "user_id": 123, "action": "login"}
{"event_id": "evt_002", "user_id": 124, "action": "view"}
{"event_id": "evt_003", "user_id": 123, "action": "purchase"}

NATS JetStream

JetStream adds durable, replayable streams to NATS while keeping the footprint tiny, which suits edge and IoT fleets where a Kafka cluster is overkill. Subjects are hierarchical, messages are opaque bytes, and one JSON document per message replays cleanly to a JSONL file.

# Create a persistent, file-backed stream
nats stream add EVENTS \
  --subjects "events.>" \
  --storage file \
  --retention limits \
  --max-age 168h \
  --dupe-window 2m

# Publish one JSON document per message
while read -r line; do nats pub events.user "$line"; done < events.jsonl

# Durable pull consumer, replayed straight back out as JSONL
nats consumer add EVENTS processor --pull --ack explicit --deliver all
nats consumer next EVENTS processor --count 100 --raw >> replay.jsonl

Redpanda Connect (formerly Benthos)

Most JSONL stream work is filter, reshape, enrich, forward. Redpanda Connect expresses that as a YAML pipeline with a mapping language called Bloblang, so a transform that would be a hundred lines of consumer code becomes a config file you can review in a pull request.

# connect.yaml - JSONL file in, filtered and enriched, out to a topic
input:
  file:
    paths: [ /data/events.jsonl ]
    scanner:
      lines: {}

pipeline:
  processors:
    # Drop noise before it costs anything downstream
    - mapping: 'root = if this.action == "heartbeat" { deleted() }'
    # Normalise and enrich every surviving line
    - mapping: |
        root = this
        root.user_id = this.user_id.string()
        root.amount = this.amount.number().catch(0)
        root.ingested_at = now()

output:
  kafka_franz:
    seed_brokers: [ localhost:9092 ]
    topic: user-events
    key: '${! json("user_id") }'

# Run it
rpk connect run connect.yaml