Quickstart

Applies to VAST Cluster 5.5.

In this quickstart you'll create a topic with a Kafka admin client, produce a few JSON events, and consume them in a consumer group. Then you'll read the same topic as a VAST DataBase table with the Python SDK, and remove what you created. You use the Confluent Kafka Python client (confluent-kafka) and the VAST DataBase SDK (vastdb). There's one path and no optional settings. It takes about 10 minutes once your platform team has set up access.

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

Before you start

You need:

A Kafka-enabled view (the broker), with a virtual IP pool, set up by your platform team. Ask them for three of the pool's virtual IPs as bootstrap servers. Port 9092 takes unencrypted connections and 9093 takes TLS. See setup for the full list of what to ask for.
If the view requires SASL/PLAIN, a VAST local user name and password. This page uses an unencrypted connection; for TLS and mTLS see security. Your platform team sets up the broker's TLS certificate and the identity policies that grant Kafka permissions (setup).
The name of the view's S3 bucket, and S3 access keys with read access to it, for the SDK part. See VAST DataBase setup prerequisites.
Python 3.10 or later.

Install the clients. VAST Cluster 5.5 supports Confluent Kafka Python client 2.4 to 2.8 (KB):

pip install "confluent-kafka>=2.4,<2.9" vastdb

Set the connection settings in your environment:

export KAFKA_BOOTSTRAP="<vip-1>:9092,<vip-2>:9092,<vip-3>:9092"
# Only if the view requires SASL/PLAIN:
export KAFKA_USERNAME="<local user>"
export KAFKA_PASSWORD="<password>"
# For the VAST DataBase SDK:
export AWS_ACCESS_KEY_ID="<access key>"
export AWS_SECRET_ACCESS_KEY="<secret key>"
export AWS_S3_ENDPOINT_URL="http://<database vip>"
export EVENTS_BUCKET="<the Kafka view's bucket>"

What your platform team sets up for a broker, in order, before your Kafka client connects. See setup for each item.

Set up

Load the settings, and check that the client can reach the broker:

import json
import os

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

TOPIC = "docs-quickstart-events"
GROUP = "docs-quickstart-group"
BUCKET = os.environ["EVENTS_BUCKET"]

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

admin = AdminClient(kafka_conf())
metadata = admin.list_topics(timeout=10)
print("connected:", len(metadata.brokers) > 0)

connected: True

Every virtual IP in the view's pool is a broker, so the metadata lists one broker per virtual IP.

Create a topic

Topics aren't created automatically, so create one before you produce to it. This one has two partitions:

futures = admin.create_topics(
    [NewTopic(TOPIC, num_partitions=2, replication_factor=1)],
    operation_timeout=30,
)
for name, future in futures.items():
    future.result()
    print("created", name)

partitions = admin.list_topics(TOPIC, timeout=10).topics[TOPIC].partitions
print("partitions:", sorted(partitions))

created docs-quickstart-events
partitions: [0, 1]

The partition count is fixed when you create the topic. See topics.

Produce events

Produce six JSON events with a key each. The producer turns off idempotence and compression, which VAST Event Broker doesn't support:

producer = Producer(kafka_conf({
    "acks": "all",
    "enable.idempotence": False,
    "compression.type": "none",
}))

delivered, failed = [], []

def on_delivery(err, msg):
    if err is not None:
        failed.append(err.name())
    else:
        delivered.append(msg.key().decode())

events = [
    {"id": 1, "user": "ana", "action": "login"},
    {"id": 2, "user": "ben", "action": "view"},
    {"id": 3, "user": "ana", "action": "purchase"},
    {"id": 4, "user": "chen", "action": "login"},
    {"id": 5, "user": "ben", "action": "logout"},
    {"id": 6, "user": "chen", "action": "view"},
]
for event in events:
    producer.produce(TOPIC, key=event["user"], value=json.dumps(event), on_delivery=on_delivery)

remaining = producer.flush(30)
print("delivered:", len(delivered), "failed:", failed, "still queued:", remaining)

delivered: 6 failed: [] still queued: 0

With acks=all, a delivery report means the events are committed to the topic. Events with the same key go to the same partition. See producing.

Consume events in a group

Read the events back in a consumer group, from the start of the topic, and commit the offsets:

consumer = Consumer(kafka_conf({
    "group.id": GROUP,
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
}))
consumer.subscribe([TOPIC])

received = []
for _ in range(60):  # poll for up to about a minute
    msg = consumer.poll(1.0)
    if msg is None:
        if len(received) == len(events):
            break
        continue
    if msg.error():
        raise KafkaException(msg.error())
    received.append(json.loads(msg.value()))

consumer.commit(asynchronous=False)
consumer.close()

for event in sorted(received, key=lambda e: e["id"]):
    print(event)

{'id': 1, 'user': 'ana', 'action': 'login'}
{'id': 2, 'user': 'ben', 'action': 'view'}
{'id': 3, 'user': 'ana', 'action': 'purchase'}
{'id': 4, 'user': 'chen', 'action': 'login'}
{'id': 5, 'user': 'ben', 'action': 'logout'}
{'id': 6, 'user': 'chen', 'action': 'view'}

The first poll in a new group waits until the group has formed, so allow a few seconds before the events arrive. Order is kept within a partition, not across partitions; the output above is sorted by id. See consuming.

Read the topic as a table

A topic is a VAST DataBase table, in the schema kafka_topics of the Kafka view's bucket. The table has the topic's name. Open it with the SDK and look at its columns:

session = vastdb.connect()

with session.transaction() as tx:
    table = tx.bucket(BUCKET).schema("kafka_topics").table(TOPIC)
    for field in table.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

Each row is one event. key and value hold the bytes the producer sent, and there are columns for the broker and producer timestamps and the headers. The last three columns are used by the cluster itself, and are empty for events from Kafka producers. See querying topics.

Read the key and value columns, and decode the JSON yourself:

with session.transaction() as tx:
    table = tx.bucket(BUCKET).schema("kafka_topics").table(TOPIC)
    rows = table.select(columns=["key", "value"]).read_all().to_pylist()

for row in sorted(rows, key=lambda r: json.loads(r["value"])["id"]):
    print(row["key"].decode(), json.loads(row["value"]))

ana {'id': 1, 'user': 'ana', 'action': 'login'}
ben {'id': 2, 'user': 'ben', 'action': 'view'}
ana {'id': 3, 'user': 'ana', 'action': 'purchase'}
chen {'id': 4, 'user': 'chen', 'action': 'login'}
ben {'id': 5, 'user': 'ben', 'action': 'logout'}
chen {'id': 6, 'user': 'chen', 'action': 'view'}

The table shows committed events only, the same ones a consumer can read. It's read-only through the VAST DataBase API: you add events by producing them. To query topics with SQL through the VAST Query Engine or Trino, or to turn JSON values into typed columns as they arrive, see query topics and blob expansion.

Clean up

Delete the consumer group and the topic:

for name, future in admin.delete_consumer_groups([GROUP], request_timeout=30).items():
    future.result()
    print("deleted group", name)

for name, future in admin.delete_topics([TOPIC], operation_timeout=30).items():
    future.result()
    print("deleted topic", name)

deleted group docs-quickstart-group
deleted topic docs-quickstart-events

Deleting a topic deletes its table and all its events.