Expand JSON events into columns

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

Blob expansion parses the JSON value of each event as it's produced, and writes the fields you choose as typed columns of a VAST DataBase table. It's new in VAST Cluster 5.5. The expansion happens in the same database transaction as the event itself, so the target table is never behind the topic (5.5.0 release notes). You then filter, join and aggregate on the fields with any VAST DataBase client, instead of parsing JSON in every query.

This guide sets up a blob expansion with the Python SDK, shows what happens to events that don't match, and changes the set of expanded fields.

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

How it works

You attach a blob expansion to a topic's value column, and point it at a target table. From then on, for each event produced to the topic:

The cluster stores the event in the topic, as usual.
It parses the value as JSON, and pulls out the fields in your expansion schema. Field names match JSON keys exactly, including case.
It writes one row to the target table, with those fields as typed columns, plus metadata columns about the event and about the parse.

Steps 1 and 3 are in one transaction: the event and its expanded row commit together or not at all. Kafka consumers of the topic see the events unchanged.

The rules (5.5.0 release notes, blobexpansion create):

JSON is the only format, and value is the only source column.
The target table must exist before you create the expansion.
A topic has one expansion, and a target table can be the target of only one expansion.
You can't rename or remove target-table columns that are part of the expansion, or rename the topic.
Events are expanded as Kafka producers write them, starting when you create the expansion. Events already in the topic aren't expanded.

Each event becomes a typed row, with its origin and diagnostics, in the same transaction as the produce.

Before you start

You need the environment variables and clients from Query topics as tables. The user of your access key also needs:

s3:TabularCreateBlobExpansion, s3:TabularAlterBlobExpansion, s3:TabularDropBlobExpansion and s3:TabularGetBlobExpansion (Required S3 Permissions for Kafka Event Parsing).
Permission to create a schema and a table in the Kafka view's bucket, and to read them.

Admin task: send to your platform team. See setup prerequisites.

Blob expansion needs VAST Cluster 5.5 and SDK 2.0.9 or later. On an older cluster, the SDK raises NotSupportedVersion ("Blob expansion requires 5.5+ VAST release").

Set up

Create a topic, a schema for the target table, and the target table. The example is a stream of orders. Create the target table with the fields you'll expand; the expansion adds its own metadata columns when you create it:

import json
import os

import pyarrow as pa
import vastdb
from confluent_kafka import Producer
from confluent_kafka.admin import AdminClient, NewTopic
from vastdb import errors
from vastdb.table import BlobExpansionConfig, ExpansionFormat

EVENTS_BUCKET = os.environ["EVENTS_BUCKET"]
TOPIC = "docs-blob-expansion-orders"
TARGET_SCHEMA = "docs_blob_expansion"
TARGET_TABLE = "orders_expanded"

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=1, replication_factor=1)])[TOPIC].result()
producer = Producer({**kafka_config(), "compression.type": "none", "enable.idempotence": False, "acks": "all"})

def produce(*values):
    for v in values:
        producer.produce(TOPIC, key="docs", value=v if isinstance(v, bytes) else json.dumps(v))
    producer.flush()

expansion_schema = pa.schema([
    ("order_id", pa.int64()),
    ("customer_id", pa.int64()),
    ("amount", pa.float64()),
    ("currency", pa.string()),
    ("status", pa.string()),
    ("event_ts", pa.string()),
])

session = vastdb.connect()
with session.transaction() as tx:
    schema = tx.bucket(EVENTS_BUCKET).create_schema(TARGET_SCHEMA, fail_if_exists=False)
    schema.create_table(TARGET_TABLE, expansion_schema, fail_if_exists=False)
print("ready")

ready

Create the target table with the fields to expand. An empty target table doesn't work with SDK 2.1.0: create_table() with an empty schema raises TypeError: 'NoneType' object is not iterable and the transaction rolls back, so nothing is created. Creating the table with the fields to expand works, because the expansion accepts columns that already exist as long as their types match.

Put the target table in a schema of its own, in the Kafka view's bucket. kafka_topics holds the topics.

Events from before the expansion

Produce one event now, before the expansion exists. You'll see later that it isn't expanded:

produce({"order_id": 1000, "customer_id": 1, "amount": 1.0, "currency": "USD", "status": "old",
         "event_ts": "2026-09-16T13:19:00Z"})
print("produced")

produced

Create the expansion

Open the topic table and call create_blob_expansion(), with the expansion schema, the target table and its schema:

with session.transaction() as tx:
    topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC)
    be = topic.create_blob_expansion(
        expansion_schema=expansion_schema,
        target_table_name=TARGET_TABLE,
        target_table_schema=TARGET_SCHEMA,
        source_column_name="value",
        config=BlobExpansionConfig(
            expansion_format=ExpansionFormat.JSON,
            copy_source_column=False,
            flatten_path=False,
            flatten_delimiter="__",
            add_missing_values_output=True,
            add_excessive_values_output=True,
        ),
    )
    print(be.source_column_name, "->", be.target_table_name)
    print([name for name, _type, _md in be.columns])

value -> /default/docs-events/docs_blob_expansion/orders_expanded
['order_id', 'customer_id', 'amount', 'currency', 'status', 'event_ts']

The SDK 2.1.0 signatures:

Table.create_blob_expansion(
    expansion_schema: pa.Schema,
    target_table_name: str,
    source_column_name: str = "value",
    config: Optional[BlobExpansionConfig] = None,
    target_table_schema: Optional[str] = None,
) -> BlobExpansion

Table.blob_expansion(source_column_name: str = "value") -> BlobExpansion

@dataclass
class BlobExpansionConfig:
    expansion_format: ExpansionFormat = ExpansionFormat.JSON
    copy_source_column: bool = False
    flatten_path: bool = False
    flatten_delimiter: str = "__"
    add_missing_values_output: bool = True
    add_excessive_values_output: bool = True

target_table_schema defaults to the topic's own schema, so pass it when the target is elsewhere. If the target table already has a column with the same name as an expanded field, the types must match. Columns it doesn't have yet are added.

The options:

Option	Effect
`copy_source_column`	Also copy the raw `value` into the target table
`flatten_path`	Turn struct fields into top-level columns named `<parent><delimiter><child>`, such as `customer__id`. Without it, structs stay nested.
`flatten_delimiter`	The delimiter for `flatten_path`. Default `__`. Field names can't contain it.
`add_missing_values_output`	Add a column that counts expected fields missing from each event
`add_excessive_values_output`	Add a column that counts JSON fields in each event that aren't in the expansion schema

Produce events

Produce eight well-formed orders. Nothing consumes the topic: the expansion writes each order into the target table as part of the produce:

clean_orders = [
    {"order_id": 2001, "customer_id": 42, "amount": 129.99, "currency": "USD", "status": "paid", "event_ts": "2026-09-16T13:20:01Z"},
    {"order_id": 2002, "customer_id": 7, "amount": 19.50, "currency": "USD", "status": "pending", "event_ts": "2026-09-16T13:20:02Z"},
    {"order_id": 2003, "customer_id": 42, "amount": 249.00, "currency": "EUR", "status": "shipped", "event_ts": "2026-09-16T13:20:03Z"},
    {"order_id": 2004, "customer_id": 88, "amount": 5.00, "currency": "USD", "status": "refunded", "event_ts": "2026-09-16T13:20:04Z"},
    {"order_id": 2005, "customer_id": 42, "amount": 74.25, "currency": "GBP", "status": "paid", "event_ts": "2026-09-16T13:20:05Z"},
    {"order_id": 2006, "customer_id": 13, "amount": 420.00, "currency": "USD", "status": "paid", "event_ts": "2026-09-16T13:20:06Z"},
    {"order_id": 2007, "customer_id": 42, "amount": 8.99, "currency": "USD", "status": "pending", "event_ts": "2026-09-16T13:20:07Z"},
    {"order_id": 2008, "customer_id": 7, "amount": 15.00, "currency": "EUR", "status": "shipped", "event_ts": "2026-09-16T13:20:08Z"},
]
produce(*clean_orders)
print("produced", len(clean_orders))

produced 8

Query the target table

The target table is an ordinary VAST DataBase table, so a filter on an expanded column is pushed down to the cluster, like any other predicate. Find one customer's orders:

with session.transaction() as tx:
    target = tx.bucket(EVENTS_BUCKET).schema(TARGET_SCHEMA).table(TARGET_TABLE)
    hits = target.select(columns=["order_id", "amount", "status"],
                         predicate=target["customer_id"] == 42).read_all()
for row in sorted(hits.to_pylist(), key=lambda r: r["order_id"]):
    print(row)

{'order_id': 2001, 'amount': 129.99, 'status': 'paid'}
{'order_id': 2003, 'amount': 249.0, 'status': 'shipped'}
{'order_id': 2005, 'amount': 74.25, 'status': 'paid'}
{'order_id': 2007, 'amount': 8.99, 'status': 'pending'}

There's one row per event produced since you created the expansion. The order with order_id 1000, from before, isn't there.

Metadata columns

Besides your fields, the cluster adds these columns to the target table:

Column	Type	Holds
`blob_expansion_key`, `blob_expansion_timestamp_broker`, `blob_expansion_timestamp_producer`, `blob_expansion_headers`	as in the topic	Copies of the event's key, timestamps and headers
`blob_expansion_value`	binary	The raw value. Only with `copy_source_column`.
`blob_expansion_kafka_partition`	int64	The event's partition
`blob_expansion_external_offset`	int64	The event's offset in its partition
`blob_expansion_error_output`	int8	`0` if the value parsed, `1` if it isn't valid JSON or isn't a JSON object
`blob_expansion_mismatches_output`	int32	Fields whose JSON type doesn't convert to the column type
`blob_expansion_missing_values_output`	int32	Fields in the expansion schema that the event doesn't have. Only with `add_missing_values_output`.
`blob_expansion_excessive_values_output`	int32	Fields in the event that aren't in the expansion schema. Only with `add_excessive_values_output`.

Don't give an expanded field one of these names: the create call fails. The partition and offset columns are the only way to query an event's partition and offset as data; the topic table itself has neither. See Query topics as tables.

What happens with bad input

Now produce five orders that don't match the expansion schema: a string where a number belongs, a missing field, two extra fields, a value that isn't valid JSON, and one with several problems at once:

bad_orders = [
    {"order_id": 3001, "customer_id": 42, "amount": "free", "currency": "USD", "status": "paid"},  # wrong type
    {"order_id": 3002, "customer_id": 9, "amount": 60.00, "currency": "USD"},                      # no status
    {"order_id": 3003, "customer_id": 9, "amount": 12.00, "currency": "USD", "status": "paid",
     "coupon": "SUMMER", "tax": 1.2},                                                            # extra fields
    b'{"order_id": 3004, "customer_id": 42, "amount":',                                            # not valid JSON
    {"order_id": 3005, "customer_id": "VIP", "amount": "lots", "extra": True},                    # several at once
]
produce(*bad_orders)

def show_target(*fields, where=None):
    """Print the chosen columns of the target rows that match where(table), a predicate run on the cluster."""
    with session.transaction() as tx:
        target = tx.bucket(EVENTS_BUCKET).schema(TARGET_SCHEMA).table(TARGET_TABLE)
        columns = list(fields) + ["blob_expansion_external_offset"]
        rows = target.select(columns=columns, predicate=where(target) if where else None).read_all()
    for row in rows.sort_by("blob_expansion_external_offset").to_pylist():
        print({k: row[k] for k in fields})

show_target("order_id", "customer_id", "status",
            "blob_expansion_error_output", "blob_expansion_mismatches_output",
            "blob_expansion_missing_values_output", "blob_expansion_excessive_values_output",
            where=lambda t: (t["order_id"] >= 3000) | t["order_id"].isnull())

{'order_id': 3001, 'customer_id': 42, 'status': 'paid', 'blob_expansion_error_output': 0, 'blob_expansion_mismatches_output': 1, 'blob_expansion_missing_values_output': 1, 'blob_expansion_excessive_values_output': 1}
{'order_id': 3002, 'customer_id': 9, 'status': None, 'blob_expansion_error_output': 0, 'blob_expansion_mismatches_output': 0, 'blob_expansion_missing_values_output': 2, 'blob_expansion_excessive_values_output': 0}
{'order_id': 3003, 'customer_id': 9, 'status': 'paid', 'blob_expansion_error_output': 0, 'blob_expansion_mismatches_output': 0, 'blob_expansion_missing_values_output': 1, 'blob_expansion_excessive_values_output': 2}
{'order_id': None, 'customer_id': None, 'status': None, 'blob_expansion_error_output': 1, 'blob_expansion_mismatches_output': 0, 'blob_expansion_missing_values_output': 0, 'blob_expansion_excessive_values_output': 0}
{'order_id': 3005, 'customer_id': None, 'status': None, 'blob_expansion_error_output': 0, 'blob_expansion_mismatches_output': 2, 'blob_expansion_missing_values_output': 3, 'blob_expansion_excessive_values_output': 3}

Every produce succeeds, including the invalid JSON. A parse problem never rejects an event: the row is written, the fields that couldn't be parsed are null, and the counters say why. The original event stays in the topic at its offset.

What each kind of problem does to the target row, as the source describes it and the orders above show:

Event	Target row
Not valid JSON, or not a JSON object (an array, a number)	Every expanded field null, `blob_expansion_error_output` = 1
A field missing	That field null; the missing-values count goes up by one per missing field. A missing struct counts each of its fields.
An extra field	Ignored; the excessive-values count goes up by one
A field of the wrong type	That field null. The mismatches count goes up by one, and so does the extra-fields count, because the field is also treated as one that didn't match the schema
A JSON `null`	Null, and not counted
An empty value	Every expanded field null, error `0`

Type conversion is strict. A JSON string doesn't convert to a number or a boolean column, so "free" in a float64 column is a mismatch. Strings do convert to string, binary, date, time, timestamp and decimal columns. JSON integers convert to integer, floating-point, date, time and timestamp columns, if the value fits. A value out of the column's range is a mismatch.

Very large values can be stored in a way the expansion can't read. Such an event's row is marked as a failed parse.

The counters add up to a health check for the stream. Sum them over the target table:

import pyarrow.compute as pc

with session.transaction() as tx:
    target = tx.bucket(EVENTS_BUCKET).schema(TARGET_SCHEMA).table(TARGET_TABLE)
    rows = target.select(columns=["blob_expansion_error_output", "blob_expansion_mismatches_output",
                                  "blob_expansion_missing_values_output",
                                  "blob_expansion_excessive_values_output"]).read_all()

errors_col = rows["blob_expansion_error_output"]
print("rows:", rows.num_rows, "with a parse error:", pc.sum(errors_col).as_py())
print("type mismatches:", pc.sum(rows["blob_expansion_mismatches_output"]).as_py(),
      "missing fields:", pc.sum(rows["blob_expansion_missing_values_output"]).as_py(),
      "extra fields:", pc.sum(rows["blob_expansion_excessive_values_output"]).as_py())

rows: 13 with a parse error: 1
type mismatches: 3 missing fields: 7 extra fields: 6

To watch parsing over time, filter on the counters and the broker timestamp, for example the events that failed to parse in the last day:

target.select(predicate=(target["blob_expansion_error_output"] == 1)
                        & (target["blob_expansion_timestamp_broker"] >= since))

The cluster also raises an alert when an event fails to parse. Blob expansion statistics are in the VAST Web UI under Analytics, as custom analytics (Overview of Kafka Event Parsing).

Add fields

When producers start sending a new field, add it to the existing expansion with add_columns(). You don't recreate the expansion or rewrite any rows. The cluster adds the columns to the target table and parses them from new events on:

with session.transaction() as tx:
    be = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC).blob_expansion("value")
    be.add_columns(columns_to_add=pa.schema([("channel", pa.string()), ("promo_code", pa.string())]))

produce(
    {"order_id": 4001, "customer_id": 42, "amount": 30.00, "currency": "USD", "status": "paid",
     "event_ts": "2026-09-16T14:00:01Z", "channel": "mobile", "promo_code": "WELCOME10"},
    {"order_id": 4002, "customer_id": 5, "amount": 45.00, "currency": "USD", "status": "paid",
     "event_ts": "2026-09-16T14:00:02Z", "channel": "web", "promo_code": "FREESHIP"},
)
show_target("order_id", "status", "channel", "promo_code", "blob_expansion_excessive_values_output",
            where=lambda t: t["order_id"].isin([2001, 3003, 4001, 4002]))

{'order_id': 2001, 'status': 'paid', 'channel': None, 'promo_code': None, 'blob_expansion_excessive_values_output': 0}
{'order_id': 3003, 'status': 'paid', 'channel': None, 'promo_code': None, 'blob_expansion_excessive_values_output': 2}
{'order_id': 4001, 'status': 'paid', 'channel': 'mobile', 'promo_code': 'WELCOME10', 'blob_expansion_excessive_values_output': 0}
{'order_id': 4002, 'status': 'paid', 'channel': 'web', 'promo_code': 'FREESHIP', 'blob_expansion_excessive_values_output': 0}

Rows from before the change have the new columns null: earlier rows aren't re-parsed. Once a field is declared, it no longer counts as an extra field. Adding a field that's already in the expansion does nothing.

add_columns() also turns on the optional outputs: add_copy_source_column, add_missing_values_output and add_excessive_values_output.

Drop fields

drop_columns() stops expanding fields. The column and its data stay in the target table, and new rows have the column null (blobexpansion drop-columns):

with session.transaction() as tx:
    be = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC).blob_expansion("value")
    be.drop_columns(columns_to_remove=pa.schema([("promo_code", pa.string())]))
    print([name for name, _type, _md in tx.bucket(EVENTS_BUCKET).schema("kafka_topics")
           .table(TOPIC).blob_expansion("value").columns])

produce({"order_id": 4003, "customer_id": 42, "amount": 12.00, "currency": "USD", "status": "paid",
         "event_ts": "2026-09-16T14:00:03Z", "channel": "web", "promo_code": "SPRING"})
show_target("order_id", "channel", "promo_code", where=lambda t: t["order_id"].isin([4001, 4003]))

['order_id', 'customer_id', 'amount', 'currency', 'status', 'event_ts', 'channel']
{'order_id': 4001, 'channel': 'mobile', 'promo_code': 'WELCOME10'}
{'order_id': 4003, 'channel': 'web', 'promo_code': None}

Dropping a field that isn't in the target table raises errors.NotFound. drop_columns() also turns the optional outputs off: remove_copy_source_column, remove_missing_values_output and remove_excessive_values_output.

Read the configuration

blob_expansion() returns the current configuration, or raises MissingBlobExpansion if the topic has none:

with session.transaction() as tx:
    be = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC).blob_expansion()
    print(be.target_table_name)
    print(be.config.copy_source_column, be.config.add_missing_values_output, be.config.add_excessive_values_output)

/default/docs-events/docs_blob_expansion/orders_expanded
False True True

be.config.flatten_path and be.config.flatten_delimiter always show their defaults: the cluster doesn't return them. Keep your own record of them.

Remove the expansion

Remove the expansion before you delete its topic: deleting a topic that still has a blob expansion fails with Unknown broker error. Drop the expansion in a transaction of its own, before you drop its target table: doing both in one transaction fails with failed to delete table handle.

drop() stops expanding new events. The target table and its rows stay (Removing Kafka Event Parsing):

with session.transaction() as tx:
    tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC).blob_expansion("value").drop()

try:
    with session.transaction() as tx:
        tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC).blob_expansion("value")
except errors.MissingBlobExpansion:
    print("no blob expansion on", TOPIC)

no blob expansion on docs-blob-expansion-orders

Throughput and replication

Expansion costs produce throughput. Parsing runs inside every produce request, so an expanded topic absorbs fewer messages per second than a plain one, and the cost grows roughly with the number of expanded columns (VAST blog). Expand the fields you query, not every field, and load-test with your real message shape.
Replicated topics. When the topic's database is replicated, the destination receives the topic and its target table like any other replicated tables. Expansion runs only where the bucket is writable, so a read-only destination doesn't parse events itself (VAST blog). The destination cluster must also support blob expansion.

Clean up

Drop the target table and its schema, then delete the topic:

with session.transaction() as tx:
    schema = tx.bucket(EVENTS_BUCKET).schema(TARGET_SCHEMA)
    schema.table(TARGET_TABLE).drop()
    schema.drop()
admin.delete_topics([TOPIC])[TOPIC].result()
print("cleaned up")

cleaned up

You can also set up blob expansion in the VAST Web UI, the VAST CLI (blobexpansion commands) or the VMS REST API; see Parsing Kafka Events into a VAST Database Table.