Formats

Ubik decides how to read a topic's bytes from the wire format, not from a setting. JSON is self-describing and needs nothing. Avro carries no field names on the wire, so it needs its schema, either a local file or a registry.

JSON

Newline-delimited JSON objects, one per message, are read with no flags. The column names and types are sampled from the payloads. This is the default the source and sink page uses throughout.

Confluent-framed JSON, the wire form a JSON Schema producer emits (a 0x00 magic byte, a big-endian schema id, then the document), is also read with no flags: the frame is stripped by its magic byte and the document decoded. A registry is not consulted, because the JSON already describes itself.

A JSON string that looks like a timestamp is typed as TIMESTAMP, so TUMBLE(event_time, ...) works directly on it.

Avro through a Schema Registry

Confluent-framed Avro (the same 0x00 + schema id frame, then raw Avro) is resolved through --schema-registry: ubik fetches the subject's schema by the id on each message and decodes it.

Register a schema for the topic's value subject, create the topic, and produce a few Avro records.

cat > payment.avsc <<'AVSC'
{"type":"record","name":"payment","fields":[
  {"name":"merchant","type":"string"},
  {"name":"amount","type":"double"},
  {"name":"event_time","type":"string"}
]}
AVSC

rpk registry schema create payments-value --schema payment.avsc --type avro
rpk topic create payments -p 3

python3 - <<'PY' | rpk topic produce payments --schema-id=topic --format '%v\n'
for w in range(2):
    ts = "2026-07-18 10:%02d:00" % w
    for merchant, amount, count in [("acme", 10.0, 3), ("globex", 3.5, 2)]:
        for _ in range(count):
            print('{"merchant":"%s","amount":%s,"event_time":"%s"}' % (merchant, amount, ts))
PY

Read it, pointing --schema-registry at the registry's REST endpoint.

ubik --from kafka://localhost:9092/payments \
     --schema-registry http://localhost:8081 \
  "SELECT merchant,
          TUMBLE(CAST(event_time AS TIMESTAMP), INTERVAL '1 MINUTE') AS w,
          count(*) AS n,
          sum(amount) AS total
   FROM payments
   GROUP BY merchant, TUMBLE(CAST(event_time AS TIMESTAMP), INTERVAL '1 MINUTE')"

{"status":{"state":"done"}}
{"stats":{"rows_in":10,"rows_out":4,"late_dropped":0,"windows_closed":2,"rows_folded":10}}
{"merchant":"acme","w":"2026-07-18 10:00:00","n":3,"total":30.0}
{"merchant":"acme","w":"2026-07-18 10:01:00","n":3,"total":30.0}
{"merchant":"globex","w":"2026-07-18 10:00:00","n":2,"total":7.0}
{"merchant":"globex","w":"2026-07-18 10:01:00","n":2,"total":7.0}

A secured registry takes credentials through UBIK_SCHEMA_REGISTRY_USER, UBIK_SCHEMA_REGISTRY_PASSWORD and UBIK_SCHEMA_REGISTRY_CA; see security.

Avro without a registry

A topic of raw Avro with no registry, every message the bare Avro body against one known schema, is read with --schema and a local .avsc. Everything else is the registry example with the flag swapped.

ubik --from kafka://localhost:9092/payments --schema payment.avsc \
  "SELECT merchant, count(*) AS n
   FROM payments
   GROUP BY merchant, TUMBLE(CAST(event_time AS TIMESTAMP), INTERVAL '1 MINUTE')"

Use --schema when the producer writes bare Avro and there is no registry to resolve; use --schema-registry when messages are framed with a schema id.

What needs what