--- title: "Expand JSON events into columns" slug: "dev-event-broker-guides-blob-expansion" description: "Learn how to set up blob expansion in VAST Cluster 5.5, parsing JSON events into typed columns for efficient querying and data management." updated: 2026-10-30T18:18:46Z published: 2026-10-30T18:18:46Z canonical: "docs.vastdata.com/dev-event-broker-guides-blob-expansion" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # 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](/docs-5-5/docs/vast-cluster-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: 1. The cluster stores the event in the topic, as usual. 2. It parses the value as JSON, and pulls out the fields in your expansion schema. Field names match JSON keys exactly, including case. 3. 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](/docs-5-5/docs/vast-cluster-5-5-0-release-notes), [blobexpansion create](/docs-5-5/docs/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. ![A JSON event value is parsed during the produce, in the same transaction, into one row of the target table. The row holds the declared fields as typed columns, copies of the event's key, timestamps and headers, the event's partition and offset, and diagnostic counters: a parse-error flag and counts of type mismatches, missing fields and extra fields. A bad event still produces a row, with the fields it couldn't parse set to null.](https://cdn.us.document360.io/89b91113-5dc6-49db-86ec-57b16dc8ce56/Images/Documentation/blob-expansion-row.svg?sv=2026-02-06&spr=https&st=2026-11-08T08%3A56%3A02Z&se=2026-11-08T09%3A56%3A02Z&sr=c&sp=r&sig=ULukwoL%2Fds4V%2BdSVu1YPW9gu5CO7lzS2tnTNUQMqatg%3D) *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](/develop/docs/dev-event-broker-guides-querying-topics#before-you-start). The user of your access key also needs: - `s3:TabularCreateBlobExpansion`, `s3:TabularAlterBlobExpansion`, `s3:TabularDropBlobExpansion` and `s3:TabularGetBlobExpansion` ([Required S3 Permissions for Kafka Event Parsing](/docs-5-5/docs/publishing-events-to-vast-event-broker#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](/develop/docs/dev-database-operations-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](/develop/docs/dev-event-broker-guides-querying-topics#columns). ### 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](/docs-5-5/docs/publishing-events-to-vast-event-broker#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](/docs-5-5/docs/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](/docs-5-5/docs/publishing-events-to-vast-event-broker#removing-kafka-event-parsing-using-the-vast-web-ui)): ``` 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](https://www.vastdata.com/blog/kafka-messages-that-arrive-already-queryable)). 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](https://www.vastdata.com/blog/kafka-messages-that-arrive-already-queryable)). 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](/docs-5-5/docs/publishing-events-to-vast-event-broker#parsing-kafka-events-into-a-vast-database-table). ## See also - [Query topics as tables](/develop/docs/dev-event-broker-guides-querying-topics): the topic table's own columns. - [Producing](/develop/docs/dev-event-broker-guides-producing): producer settings for VAST Event Broker. - [Limits](/develop/docs/dev-event-broker-reference-limits). - [Querying](/develop/docs/dev-database-guides-querying) in the VAST DataBase docs: filter and tune queries on the target table.