Perspective with Kafka and other message queues

Perspective has no queue-specific connector because it does not need one: a Table is updated by calling update(), so any consumer loop is an integration. The pattern is the same for Kafka, Redpanda, NATS, RabbitMQ, Redis streams or a WebSocket feed.

1. Create a Table from a schema, on a perspective.Server.
2. Consume messages, batch them, and call table.update(batch).
3. Host the server on a WebSocket so browsers can open the table.

Python

import json
import threading

import tornado.ioloop
import tornado.web
from confluent_kafka import Consumer
from perspective import Server
from perspective.handlers.tornado import PerspectiveTornadoHandler

server = Server()
client = server.new_local_client()
table = client.table(
    {"order_id": "string", "symbol": "string", "qty": "integer", "price": "float", "ts": "datetime"},
    index="order_id",
    name="orders",
)

def consume():
    consumer = Consumer({"bootstrap.servers": "localhost:9092", "group.id": "perspective"})
    consumer.subscribe(["orders"])
    while True:
        messages = consumer.consume(num_messages=500, timeout=0.1)
        rows = [json.loads(m.value()) for m in messages if m.error() is None]
        if rows:
            table.update(rows)

threading.Thread(target=consume, daemon=True).start()

app = tornado.web.Application([
    (r"/websocket", PerspectiveTornadoHandler, {"perspective_server": server}),
])

app.listen(8080)
tornado.ioloop.IOLoop.current().start()

confluent_kafka is used here for illustration; nothing in Perspective depends on it.

Design notes

• Batch. One update() of 500 rows is much cheaper than 500 updates of one row. Consume in small time windows, as above.
• Pick index or limit. A topic is unbounded; a browser is not. Use index when messages are upserts to entities (orders, positions, devices), and limit when they are events and you want the most recent n.
• Threads are fine. Perspective’s Python API is thread-safe and releases the GIL. See Multithreading.
• Arrow if you have it. If your messages are already Arrow record batches, pass the bytes straight to update().

The browser side is identical to any other server-hosted table; see Real-time dashboards over WebSocket.