Avro streams

An Avro stream file holds records shaped like streamed data: each record names one channel, with its own timestamps, values and tags. Upload one with add_avro_stream().

Connect the client

This guide assumes you have installed the client and setup your API keys.

from nominal.core import NominalClient

client = NominalClient.from_profile("default")

When to use Avro streams

Use Parquet when every channel shares one timestamp column. Use an Avro stream when they don’t:

	Parquet	Avro stream
Timestamps	One column for every channel	Per record, so channels can sample at different times
Sparse channels	A cell for every channel in every row	Only the samples that exist
Tags	One column per tag, the same for the whole file	A map on each record
Layout	Columnar, compresses well	One record per channel batch

Because the records match the streaming format, a program that streams can also write them to disk and upload them later.

Write a file

nominal_streaming, installed with nominal on supported platforms, writes Avro streams with the same enqueue calls used to stream. Writing to a file needs no API key:

from datetime import datetime, timezone
from pathlib import Path

from nominal_streaming import NominalDatasetStream

with NominalDatasetStream().to_file(Path("flight.avro")) as stream:
    stream.enqueue("altitude", datetime.now(timezone.utc), 1250.5, tags={"vehicle_id": "sn-001"})
    stream.enqueue_batch("airspeed", [1_757_500_000_000_000_000, 1_757_500_100_000_000_000], [42.0, 42.5])
    stream.enqueue_from_dict(1_757_500_200_000_000_000, {"altitude": 1252.0, "fault_code": "E07"})

Batch timestamps are epoch nanoseconds.

To keep data when a live stream loses its connection, pass file_fallback to get_write_stream(). Batches that fail to send are written to that .avro file instead:

dataset = client.get_dataset("<DATASET_RID>")
with dataset.get_write_stream(implementation="rust", file_fallback="flight_backup.avro") as stream:
    stream.enqueue("altitude", datetime.now(timezone.utc), 1250.5)

file_fallback needs the Rust stream. The Python stream logs a warning and drops failed batches.

Upload a file

dataset = client.create_dataset("Flight telemetry")
file = dataset.add_avro_stream("flight.avro", tags={"vehicle_id": "sn-001"})
file.poll_until_ingestion_completed()

timestamp_type defaults to "epoch_nanoseconds". It takes numeric types only: another epoch unit, or ts.Relative. tags fill in keys that a record doesn’t set. Tags in the records take precedence.

Asset and Run also have add_avro_stream, which uploads to a dataset by data scope name or ref name. To include an Avro stream in a batch, see Ingest jobs.

Value types

Each value is a double, long, string, double array, string array or JSON struct.

• Every value for one channel and tag set must use the same type. Mixing them fails ingestion.

• The type sets the channel’s data type. A long value for an existing DOUBLE channel creates a second series. Register channels to avoid this.

• Booleans, datetimes and decimals have no type. Convert them first, for example a boolean to double or string.

• Integer arrays become double arrays.

• NaN and infinity are valid double values, and NaN is the usual way to mark a missing sample. JSON structs can’t hold them, so replace them with null first.

Schema

Writers in other languages need this schema. The order of the types in the values union must match:

{
  "type": "record",
  "name": "AvroStream",
  "namespace": "io.nominal.ingest",
  "fields": [
    {"name": "channel", "type": "string"},
    {"name": "timestamps", "type": {"type": "array", "items": "long"}},
    {"name": "values", "type": {"type": "array", "items": [
      "double",
      "long",
      "string",
      {"type": "record", "name": "DoubleArray",
       "fields": [{"name": "items", "type": {"type": "array", "items": "double"}}]},
      {"type": "record", "name": "StringArray",
       "fields": [{"name": "items", "type": {"type": "array", "items": "string"}}]},
      {"type": "record", "name": "JsonStruct",
       "fields": [{"name": "json", "type": "string"}]}
    ]}},
    {"name": "tags", "type": {"type": "map", "values": "string"}, "default": {}}
  ]
}

timestamps and values are parallel arrays of the same length. A file that doesn’t match the schema fails ingestion. The earlier schema, with only double and string values, is still accepted.