Load JSON, CSV and Parquet files

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

This guide loads the three most common file formats into VAST DataBase tables with the Python SDK and PyArrow: newline-delimited JSON, CSV and Parquet. For each one it shows the type problems you'll meet when you let PyArrow guess the schema, how to give it the schema instead, and how to stream a large file in batches so it never has to fit in memory. For the loading methods themselves, and when to use import_files() instead, see ingest.

Run the code blocks in order in one Python session. Below each block is the output it produced on a VAST Cluster 5.5 system.

Before you start

You need the connection settings from the quickstart in your environment, and PyArrow, which the SDK installs. Connect, and create the schema for this guide:

import io
import os
import tempfile

import pyarrow as pa
import pyarrow.csv as pa_csv
import pyarrow.json as pa_json
import pyarrow.parquet as pq
import vastdb
from vastdb import errors

session = vastdb.connect()
BUCKET = os.environ["VASTDB_BUCKET"]
SCHEMA = "guide_load_files"
workdir = tempfile.mkdtemp()

with session.transaction() as tx:
    tx.bucket(BUCKET).create_schema(SCHEMA, fail_if_exists=False)
print("Schema ready:", SCHEMA)

Schema ready: guide_load_files

All three formats end the same way: a stream of Arrow record batches, inserted into a table. This helper creates the table if it doesn't exist, and inserts each batch in its own transaction:

def load(table_name, schema, batches):
    with session.transaction() as tx:
        tx.bucket(BUCKET).schema(SCHEMA).create_table(table_name, schema, fail_if_exists=False)
    rows = requests = 0
    for batch in batches:
        with session.transaction() as tx:
            tx.bucket(BUCKET).schema(SCHEMA).table(table_name).insert(batch)
        rows += batch.num_rows
        requests += 1
    return rows, requests

def rows_with(table_name, columns, key, values):
    """Read only the rows whose key is in values: the filter runs on the cluster."""
    with session.transaction() as tx:
        t = tx.bucket(BUCKET).schema(SCHEMA).table(table_name)
        rows = t.select(columns=columns, predicate=t[key].isin(values)).read_all()
    return rows.sort_by(key).to_pylist()

rows_with() reads back a couple of rows to check a load. It passes a predicate, so only those rows come back; reading the whole table to look at two rows would send every row to your client. (limit_rows stops a query early, but which rows it returns isn't defined; see querying.)

One transaction per batch means a failure partway through leaves the batches before it loaded. If you need all or nothing, insert every batch in one transaction instead; see transactions in practice. For how big a batch should be, see ingest.

JSON

PyArrow reads newline-delimited JSON: one JSON object per line. (For a file that is one big JSON array, load it with json.load() and build the table with pa.Table.from_pylist().) Write a sample file of orders:

json_path = os.path.join(workdir, "orders.ndjson")
with open(json_path, "w") as f:
    for i in range(1, 20001):
        f.write(f'{{"order_id": {i}, "ts": "2026-09-16T13:{i % 60:02d}:00", '
                f'"amount": {i % 50 + 0.5}, "coupon": null}}\n')
print(os.path.getsize(json_path), "bytes")

1604894 bytes

What PyArrow guesses

Let PyArrow infer the schema, and try to create a table from it:

inferred = pa_json.read_json(json_path).schema
print(inferred)

try:
    with session.transaction() as tx:
        tx.bucket(BUCKET).schema(SCHEMA).create_table("orders_inferred", inferred)
except errors.NotSupportedSchema as e:
    print(type(e).__name__, e.cause.code)
    print(e.message)

order_id: int64
ts: timestamp[s]
amount: double
coupon: null
NotSupportedSchema TabularUnsupportedColumnType
The field named 'coupon' has an unsupported Arrow type 'null'

coupon is null in every line, so PyArrow gives it the null type, and the cluster rejects a null column. The SDK raises NotSupportedSchema; see errors. The same happens to any field that's null in the part of the file PyArrow looks at. Inference has other traps too: a field that's an integer in some lines and a decimal in others, or a timestamp that isn't in ISO format and stays a string.

Give PyArrow the schema

Declare the schema you want. PyArrow then parses each field as that type, and converts ISO timestamp strings for you. unexpected_field_behavior="ignore" drops fields you didn't declare:

orders_schema = pa.schema([
    ("order_id", pa.int64()),
    ("ts", pa.timestamp("s")),
    ("amount", pa.float64()),
    ("coupon", pa.string()),
])

reader = pa_json.open_json(
    json_path,
    read_options=pa_json.ReadOptions(block_size=256 * 1024),
    parse_options=pa_json.ParseOptions(explicit_schema=orders_schema,
                                       unexpected_field_behavior="ignore"),
)
rows, requests = load("orders", orders_schema, reader)
print("loaded", rows, "rows in", requests, "batches")
print(rows_with("orders", ["order_id", "ts", "amount", "coupon"], "order_id", [1, 2]))

loaded 20000 rows in 7 batches
[{'order_id': 1, 'ts': datetime.datetime(2026, 9, 16, 13, 1), 'amount': 1.5, 'coupon': None}, {'order_id': 2, 'ts': datetime.datetime(2026, 9, 16, 13, 2), 'amount': 2.5, 'coupon': None}]

open_json() reads the file a block at a time and yields record batches, so memory use stays flat however big the file is. block_size sets how much it reads at once, and so roughly how big each batch is.

CSV

Write a sample CSV of stores. The zip codes have leading zeros, and the last-visit times aren't in ISO format:

csv_path = os.path.join(workdir, "stores.csv")
with open(csv_path, "w") as f:
    f.write("store_id,zip,opened,last_visit\n")
    for i in range(1, 5001):
        f.write(f"{i},{i % 100:05d},2026-01-{i % 28 + 1:02d},16/09/2026 {i % 24:02d}:30\n")
print(open(csv_path).read().splitlines()[:3])

['store_id,zip,opened,last_visit', '1,00001,2026-01-02,16/09/2026 01:30', '2,00002,2026-01-03,16/09/2026 02:30']

What PyArrow guesses

guessed = pa_csv.read_csv(csv_path)
print(guessed.schema)
print(guessed.slice(0, 2).to_pylist())

store_id: int64
zip: int64
opened: date32[day]
last_visit: string
[{'store_id': 1, 'zip': 1, 'opened': datetime.date(2026, 1, 2), 'last_visit': '16/09/2026 01:30'}, {'store_id': 2, 'zip': 2, 'opened': datetime.date(2026, 1, 3), 'last_visit': '16/09/2026 02:30'}]

The zip codes look like numbers, so PyArrow reads them as int64, and 00001 becomes 1: the leading zeros are gone. The last-visit times aren't in a format PyArrow recognises, so they stay strings.

Give PyArrow the types

ConvertOptions sets a column's type, and timestamp_parsers tells PyArrow how to read your timestamp format. open_csv() streams the file in blocks, like open_json():

stores_schema = pa.schema([
    ("store_id", pa.int64()),
    ("zip", pa.string()),
    ("opened", pa.date32()),
    ("last_visit", pa.timestamp("s")),
])

reader = pa_csv.open_csv(
    csv_path,
    read_options=pa_csv.ReadOptions(block_size=64 * 1024),
    convert_options=pa_csv.ConvertOptions(column_types=stores_schema,
                                          timestamp_parsers=["%d/%m/%Y %H:%M"]),
)
rows, requests = load("stores", stores_schema, reader)
print("loaded", rows, "rows in", requests, "batches")
print(rows_with("stores", ["store_id", "zip", "opened", "last_visit"], "store_id", [1, 2]))

loaded 5000 rows in 3 batches
[{'store_id': 1, 'zip': '00001', 'opened': datetime.date(2026, 1, 2), 'last_visit': datetime.datetime(2026, 9, 16, 1, 30)}, {'store_id': 2, 'zip': '00002', 'opened': datetime.date(2026, 1, 3), 'last_visit': datetime.datetime(2026, 9, 16, 2, 30)}]

Other ConvertOptions you'll often need: null_values (which strings mean null, for example ["", "NA"]), true_values and false_values for booleans, and include_columns to load only some columns. Set ReadOptions(column_names=...) when the file has no header row.

Parquet

A Parquet file already has a schema, so there's nothing to guess. Two questions decide how you load it:

Check the types

Parquet files written by other tools often hold Arrow types that the cluster doesn't accept: large_string, dictionary-encoded columns, durations, and null columns. Write a sample file with all four:

parquet_path = os.path.join(workdir, "readings.parquet")
n = 10000
sample = pa.table({
    "reading_id": pa.array(range(n), pa.int64()),
    "sensor": pa.array([f"sensor-{i % 7}" for i in range(n)]).dictionary_encode(),
    "site": pa.array([f"site-{i % 3}" for i in range(n)], pa.large_string()),
    "elapsed": pa.array([i * 1_000_000_000 for i in range(n)], pa.duration("ns")),
    "note": pa.nulls(n),
})
pq.write_table(sample, parquet_path)
print(pq.ParquetFile(parquet_path).schema_arrow)

reading_id: int64
sensor: dictionary<values=string, indices=int32, ordered=0>
site: large_string
elapsed: duration[ns]
note: null

This function lists the columns whose types the cluster rejects, with a type to use instead. The list is from data types:

def check_schema(schema):
    fixes = {}
    for field in schema:
        t = field.type
        if pa.types.is_large_string(t) or pa.types.is_string_view(t):
            fixes[field.name] = pa.string()
        elif pa.types.is_large_binary(t):
            fixes[field.name] = pa.binary()
        elif pa.types.is_dictionary(t):
            fixes[field.name] = t.value_type
        elif pa.types.is_date64(t):
            fixes[field.name] = pa.date32()
        elif pa.types.is_duration(t):
            fixes[field.name] = pa.int64()
        elif pa.types.is_null(t):
            fixes[field.name] = None  # no data to go on: you choose the type
    return fixes

fixes = check_schema(pq.ParquetFile(parquet_path).schema_arrow)
for name, fix in fixes.items():
    print(f"{name}: change to {fix}" if fix is not None else f"{name}: all null, choose a type")

sensor: change to string
site: change to string
elapsed: change to int64
note: all null, choose a type

A duration converts to an integer in its own unit, so elapsed becomes nanoseconds as an int64. Rename the column, or convert it to seconds, so the unit stays obvious. A null column has no data to suggest a type, so you pick one.

Load in batches

Build the target schema from the fixes, then read the file a batch at a time and cast each batch to it. iter_batches() reads one batch at a time, so the file never has to fit in memory:

fixes["note"] = pa.string()
source = pq.ParquetFile(parquet_path)
target_schema = pa.schema([pa.field(f.name, fixes.get(f.name, f.type)) for f in source.schema_arrow])
print(target_schema)

batches = (batch.cast(target_schema) for batch in source.iter_batches(batch_size=4000))
rows, requests = load("readings", target_schema, batches)
print("loaded", rows, "rows in", requests, "batches")
print(rows_with("readings", ["reading_id", "sensor", "site", "elapsed", "note"], "reading_id", [0, 1]))

reading_id: int64
sensor: string
site: string
elapsed: int64
note: string
loaded 10000 rows in 3 batches
[{'reading_id': 0, 'sensor': 'sensor-0', 'site': 'site-0', 'elapsed': 0, 'note': None}, {'reading_id': 1, 'sensor': 'sensor-1', 'site': 'site-1', 'elapsed': 1000000000, 'note': None}]

To fix a file and still let the cluster import it, write the fixed batches to a new Parquet file in a bucket on the cluster with pq.ParquetWriter, and import that with import_files().

Clean up

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema(SCHEMA)
    for table in schema.tables():
        table.drop()
    schema.drop()
print("Dropped", SCHEMA)

Dropped guide_load_files

See also

• Ingest: insert(), import_files(), batch sizes, and new columns in incoming data.
• Load scientific array data: NetCDF and GRIB2.
• Data types: every type the cluster accepts, and what to use instead of the rest.
• Data modeling: choose a table structure before a large load.