Query topics as tables

Applies to VAST Cluster 5.5. Examples use the VAST DataBase Python SDK 2.1.0.

Every VAST Event Broker topic is also a VAST DataBase table. You can read the events that Kafka producers wrote with the VAST DataBase Python SDK, with SQL through the VAST Query Engine, or with Trino and Spark, without a Kafka consumer. This guide shows where the tables are, what their columns hold, and how to read them. It also covers what you can't do with them.

Run the code blocks in order in one Python session. Below each block is the output you should see.

Where topics live

When your platform team creates a Kafka view, the cluster creates a VAST DataBase database for it: the view's bucket. Each topic is a table in that bucket's kafka_topics schema, with the same name as the topic (Managing Event Topics). So topic orders on a Kafka view with bucket events is the table:

Each partition is stored separately inside the table, but you query the table as a whole.

The same topic, read by a Kafka consumer or queried as a table.

Columns

Each row of a topic table is one event. The columns are:

key and value are raw bytes. The cluster doesn't parse them, so decode them in your code. To get typed columns from JSON values, set up blob expansion.

There's no partition column and no offset column: you can't query a topic by partition (Kafka Protocol Support). The cluster uses an event's row ID in its partition as the Kafka offset, but the row ID alone doesn't tell you the partition. If you need the partition and offset as columns, blob expansion adds them to its target table.

Before you start

You need:

• A topic to read, and the bucket of its Kafka view. This guide creates its own topic with a Kafka client; see Topics.
• For the Kafka client: the bootstrap servers, and credentials if the view uses SASL/PLAIN. See Setup.
• For the SDK: an S3 endpoint and an access key pair for a user with read access (the s3:Tabular* read actions) on the Kafka view's bucket. See setup prerequisites.
• For SQL: a Query Engine endpoint and s3:QueryEngine*. See VAST Query Engine (ADBC).

Set these environment variables:

export KAFKA_BOOTSTRAP="<vip>:9092"
export KAFKA_USERNAME="<VAST local user>"      # only if the view uses SASL/PLAIN
export KAFKA_PASSWORD="<password>"             # only if the view uses SASL/PLAIN
export AWS_S3_ENDPOINT_URL="http://<endpoint>"
export AWS_ACCESS_KEY_ID="<access key>"
export AWS_SECRET_ACCESS_KEY="<secret key>"
export EVENTS_BUCKET="<bucket of the Kafka view>"
export QE_ENDPOINT="http://<query engine vip>"

Install the clients:

pip install "vastdb[adbc]==2.1.0" "confluent-kafka>=2.4,<2.9"

Set up

Create a topic, produce three JSON events to it, and connect the SDK:

import json
import os
from datetime import datetime, timedelta, timezone

import vastdb
from confluent_kafka import Producer
from confluent_kafka.admin import AdminClient, NewTopic

EVENTS_BUCKET = os.environ["EVENTS_BUCKET"]
TOPIC = "docs-querying-topics-orders"

def kafka_config():
    cfg = {"bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"]}
    if os.environ.get("KAFKA_USERNAME"):
        cfg.update({
            "security.protocol": "SASL_PLAINTEXT",
            "sasl.mechanism": "PLAIN",
            "sasl.username": os.environ["KAFKA_USERNAME"],
            "sasl.password": os.environ["KAFKA_PASSWORD"],
        })
    return cfg

admin = AdminClient(kafka_config())
admin.create_topics([NewTopic(TOPIC, num_partitions=3, replication_factor=1)])[TOPIC].result()

producer = Producer({**kafka_config(), "compression.type": "none", "enable.idempotence": False, "acks": "all"})
orders = [
    {"order_id": 1, "status": "new", "amount": 19.5},
    {"order_id": 2, "status": "new", "amount": 7.25},
    {"order_id": 1, "status": "paid", "amount": 19.5},
]
for order in orders:
    producer.produce(TOPIC, key=f"order-{order['order_id']}", value=json.dumps(order),
                     headers=[("source", b"docs")])
producer.flush()

session = vastdb.connect()
print("ready")

ready

flush() returns once the broker has acknowledged every record. The broker acknowledges a produce request after its events are committed, so they're readable as table rows straight away.

Read with the Python SDK

Open the topic table like any other table, and look at its columns:

with session.transaction() as tx:
    topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC)
    for field in topic.columns():
        print(field.name, field.type)

key binary
value binary
timestamp_broker timestamp[ms]
timestamp_producer timestamp[ms]
headers list<item: struct<key: string not null, value: binary> not null>
external_offset int64
compaction_process_seq_num int64
extra_metadata binary

Read the events and decode the payloads. As with any SDK query, call read_all() before the transaction ends:

with session.transaction() as tx:
    topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC)
    events = topic.select(columns=["key", "value", "timestamp_broker", "headers"]).read_all()

for row in sorted(events.to_pylist(), key=lambda r: (r["key"], r["value"])):
    print(row["key"].decode(), json.loads(row["value"]), [(h["key"], h["value"]) for h in row["headers"]])

order-1 {'order_id': 1, 'status': 'new', 'amount': 19.5} [('source', b'docs')]
order-1 {'order_id': 1, 'status': 'paid', 'amount': 19.5} [('source', b'docs')]
order-2 {'order_id': 2, 'status': 'new', 'amount': 7.25} [('source', b'docs')]

Rows don't come back in any guaranteed order: the cluster reads the table partition by partition. Sort on the client if order matters.

Filter on the cluster

Filters and column lists are applied on the cluster, so only matching events come back. A common filter is a time window on timestamp_broker:

since = datetime.now(timezone.utc) - timedelta(hours=1)

with session.transaction() as tx:
    topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC)
    recent = topic.select(columns=["key", "value"], predicate=topic["timestamp_broker"] >= since).read_all()

print(recent.num_rows, "events in the last hour")

3 events in the last hour

You can't filter on fields inside value, because the cluster sees it as bytes. To filter or aggregate on payload fields, use blob expansion and query its target table. The DB guide Querying covers predicates and tuning.

Query with the VAST Query Engine

The Query Engine runs SQL on the cluster. Connect with the ADBC driver, and name the table as "<bucket>/kafka_topics"."<topic>". Quote the topic name, since topic names often contain - or .:

import adbc_driver_manager.dbapi as dbapi
import adbc_driver_vastdb

conn = dbapi.connect(
    driver=adbc_driver_vastdb.get_driver_path(),
    db_kwargs={
        "vast.db.endpoint": os.environ["QE_ENDPOINT"],
        "vast.db.access_key": os.environ["AWS_ACCESS_KEY_ID"],
        "vast.db.secret_key": os.environ["AWS_SECRET_ACCESS_KEY"],
    },
    autocommit=True,
)

TABLE = f'"{EVENTS_BUCKET}/kafka_topics"."{TOPIC}"'
with conn.cursor() as cur:
    cur.execute(f"SELECT count(*) AS events FROM {TABLE} WHERE key = 'order-1'::BLOB")
    print("events for order-1:", cur.fetchone()[0])

    cur.execute(f"SELECT key, value FROM {TABLE} ORDER BY value DESC LIMIT 2")
    for k, value in cur.fetchall():
        print(k.decode(), value.decode())

events for order-1: 2
order-2 {"order_id": 2, "status": "new", "amount": 7.25}
order-1 {"order_id": 1, "status": "paid", "amount": 19.5}

The aggregation and the filter run on the cluster. key and value are binary, so compare them with a binary value: 'order-1'::BLOB, or CAST(key AS VARCHAR) = 'order-1'. Don't use the SQL-standard hex literal X'6f72…': it runs without an error but matches nothing. The example decodes the results in Python.

In VAST Cluster 5.5 the Query Engine doesn't support GROUP BY, and accepts ORDER BY only with a LIMIT: GROUP BY fails with "The operator 'HASH_GROUP_BY' ... is currently not supported", and ORDER BY without LIMIT with "The operator 'ORDER_BY' ... is currently not supported for this usage". To group events, use Trino or Spark, or read the columns with the SDK and group them with PyArrow (table.group_by()). See VAST Query Engine (ADBC) for the driver, and the SQL reference for the SQL you can use.

Query with Trino or Spark

The VAST DataBase connectors for Trino and Spark see topic tables like any other table. In Trino the topics schema is <bucket>/kafka_topics: for example kafka2/kafka_topics for a Kafka view whose bucket is kafka2 (Querying Events from Topics using VAST Database Connectors). In Trino, decode the binary payload with from_utf8(), and read JSON fields with the JSON functions:

SELECT from_utf8(key) AS order_key,
       json_extract_scalar(from_utf8(value), '$.status') AS status,
       timestamp_broker
FROM vast."events/kafka_topics"."docs-querying-topics-orders"
ORDER BY timestamp_broker;

Parsing JSON in every query costs CPU on the Trino workers. For payload fields you query often, blob expansion gives you typed columns that the cluster can filter. See Trino and Spark to set up the connectors.

What you can't do with a topic table

Topic tables are read-only through the VAST DataBase API. You write events with a Kafka producer. The cluster refuses inserts, updates, deletes, imports and column changes on a topic table:

import pyarrow as pa

try:
    with session.transaction() as tx:
        topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC)
        topic.insert(pa.table({"key": [b"order-9"], "value": [b"{}"]}))
except Exception as e:
    print(type(e).__name__)

BadRequest

Also:

• You can't create a topic through the VAST DataBase API. Create it with a Kafka client or VMS; see Topics.
• If a topic's maximum message size is set above 5 MB, database queries on the topic aren't supported (Creating a Topic in VAST Web UI).
• Retention and compaction remove rows in the background, so a query only sees what the topic still holds. See Topics.

Warning: dropping a topic table deletes the topic. table.drop() in the SDK, or DROP TABLE in SQL, on a table in kafka_topics deletes the Kafka topic and all its events, as if you'd deleted it with a Kafka admin client. Producers and consumers of that topic start getting errors.

Reads see committed events only

A query sees the events of produce requests that have committed, and none of a request that's still in progress. That's the same view a Kafka consumer gets. Each Produce request is one database transaction, so a query never sees part of one. See architecture.

Clean up

Delete the topic with the Kafka admin client, and close the connections:

conn.close()
admin.delete_topics([TOPIC])[TOPIC].result()
print("cleaned up")

cleaned up

See also

• Blob expansion: turn JSON values into typed columns as events arrive.
• Consuming: read the same events with a Kafka consumer.
• Architecture: how topics are stored as tables.
• Querying, VAST Query Engine (ADBC) and Trino in the VAST DataBase docs.