Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Your First Pipeline: Kafka To Redis

The first graph is deliberately trivial: JSON order records arrive on a Kafka topic, Nervix decodes them, and an emitter publishes them to a Redis Pub/Sub channel. Along the way you meet every building block that later steps reuse: domains, schemas, codecs, relays, clients, ingestors, and emitters.

Provision The Topic

Nervix never creates external entities, so create the Kafka topic first:

docker exec broker /opt/kafka/bin/kafka-topics.sh \
  --create --topic orders --bootstrap-server localhost:9092

Create A Domain

Every entity in Nervix lives in an explicit domain — there is no implicit default. Open nervix-cli and create one, then switch the session to it:

CREATE UNPACED DOMAIN quickstart;

USE quickstart;

An unpaced domain admits records as they arrive, with no domain clock pacing — the right choice for a first pipeline. See Domains And Time. USE is a client-local statement that scopes the rest of your session to the domain.

Open A Transaction

Submitting more than one model statement at a time requires a transaction, and it is also how a graph is assembled atomically — nothing takes effect until COMMIT;. See NSPL Overview for the statement surface.

BEGIN;

The graph declarations in the rest of this page run inside this transaction.

Describe The Data

Nervix separates what a record looks like inside the graph from how it is serialized on the wire. The internal schema is exact-typed; the wire schema describes the JSON payload in the wire format’s own type names; the codec bridges the two. See Schemas And Codecs.

CREATE SCHEMA order_record (
  order_id STRING,
  customer STRING,
  status STRING,
  amount I64,
  quantity I64
);

CREATE WIRE JSON SCHEMA order_wire MODE STRICT (
  order_id string,
  customer string,
  status string,
  amount integer,
  quantity integer
);

CREATE CODEC order_codec
  FROM WIRE JSON SCHEMA order_wire
  TO SCHEMA order_record;

STRICT rejects payload fields that are not declared; LOOSE would drop them instead (Wire Schemas). There are no implicit casts anywhere in Nervix, so the codec is the only place where wire values become typed internal values. A DATETIME field would additionally need an explicit encoding such as ENCODE created_at AS RFC3339 (Codecs).

Create A Relay

Relays are the named streams that connect runtime nodes. Each relay carries records of exactly one schema. See Streams And State.

CREATE RELAY orders SCHEMA order_record UNBRANCHED;

UNBRANCHED means one shared stream. Branched Processing introduces the alternative.

Declare The External Connections

Connection endpoints never live on ingestors or emitters; they are named CLIENT objects that nodes reference. Kafka client config is passed through to the Kafka driver, so bootstrap.servers uses its familiar key.

CREATE CLIENT kafka_local
  TYPE KAFKA
  CONFIG {
    'bootstrap.servers' = '127.0.0.1:9092'
  };

CREATE CLIENT redis_local
  TYPE REDIS
  POOL SIZE MIN 1 MAX 4
  CONFIG {
    'addr' = 'redis://127.0.0.1:6379/'
  };

Ingest From Kafka

An ingestor brings external data into the graph. It decodes each payload with the codec and routes the result to a relay:

CREATE KAFKA TOPIC INGESTOR kafka_orders
  FROM kafka_local
  TOPIC 'orders'
  OFFSET BY CONSUMER GROUP 'quickstart'
  MODE ACK SEQUENTIAL ACK TIMEOUT 30s RETRY POLICY BACKOFF 200ms MAX 5s
  ON QUIESCE SUSPEND DECODE USING order_codec
  TO orders
    INHERIT ALL
    UNBRANCHED
    FLUSH EACH 100ms MAX BATCH SIZE 1MiB
    ON MESSAGE ERROR LOG
  ON GENERAL ERROR LOG;

Reading top to bottom:

  • KAFKA TOPIC INGESTOR names the connector and the entity the ingestor reads. FROM names the TYPE KAFKA client, and the topic and consumer group are string literals sent to Kafka exactly as written.
  • OFFSET BY CONSUMER GROUP tracks progress in Kafka; MODE ACK SEQUENTIAL acknowledges records in order. The full Kafka source grammar is in Supported Ingestor Types.
  • TO orders starts a route. Routes begin empty: INHERIT ALL copies every decoded field into the outgoing record, and later steps show SET constructing new values. See The Working Message.
  • Every route needs an explicit flush policy — FLUSH EACH <duration> MAX BATCH SIZE <bytes> or FLUSH IMMEDIATE. There are no hidden defaults.
  • ON MESSAGE ERROR LOG is the route’s per-record error policy; ON GENERAL ERROR LOG is the node-wide policy (Runtime Node Error Policies).

Emit To Redis

An emitter drains a relay out of the graph. The Redis integration is Pub/Sub (Supported Emitter Sinks), and the same codec that decoded the input encodes it back to JSON:

CREATE EMITTER redis_orders
  FROM orders
  TO REDIS PUBSUB redis_local CHANNEL orders_out
    MODE NO_ACK RETRY POLICY BACKOFF 250ms MAX 30s
    ENCODE USING order_codec
  INHERIT ALL
  FLUSH EACH 100ms MAX BATCH SIZE 1MiB
  ON MESSAGE ERROR LOG
  ON GENERAL ERROR LOG;

Redis Pub/Sub exposes no delivery confirmation, so MODE NO_ACK names server acceptance as the publish boundary. The required retry policy paces connection and other infrastructure failures; it does not turn subscriber presence into an acknowledgment.

Commit And Start

COMMIT; applies the structural changes: the whole graph is validated and installed atomically. Nothing processes data yet — that is the job of START;, the domain lifecycle command that begins actual work (Start And Stop):

COMMIT;

START;

The Complete Script

CREATE UNPACED DOMAIN quickstart;

USE quickstart;

BEGIN;

CREATE SCHEMA order_record (
  order_id STRING,
  customer STRING,
  status STRING,
  amount I64,
  quantity I64
);

CREATE WIRE JSON SCHEMA order_wire MODE STRICT (
  order_id string,
  customer string,
  status string,
  amount integer,
  quantity integer
);

CREATE CODEC order_codec
  FROM WIRE JSON SCHEMA order_wire
  TO SCHEMA order_record;

CREATE RELAY orders SCHEMA order_record UNBRANCHED;

CREATE CLIENT kafka_local
  TYPE KAFKA
  CONFIG {
    'bootstrap.servers' = '127.0.0.1:9092'
  };

CREATE CLIENT redis_local
  TYPE REDIS
  POOL SIZE MIN 1 MAX 4
  CONFIG {
    'addr' = 'redis://127.0.0.1:6379/'
  };

CREATE KAFKA TOPIC INGESTOR kafka_orders
  FROM kafka_local
  TOPIC 'orders'
  OFFSET BY CONSUMER GROUP 'quickstart'
  MODE ACK SEQUENTIAL ACK TIMEOUT 30s RETRY POLICY BACKOFF 200ms MAX 5s
  ON QUIESCE SUSPEND DECODE USING order_codec
  TO orders
    INHERIT ALL
    UNBRANCHED
    FLUSH EACH 100ms MAX BATCH SIZE 1MiB
    ON MESSAGE ERROR LOG
  ON GENERAL ERROR LOG;

CREATE EMITTER redis_orders
  FROM orders
  TO REDIS PUBSUB redis_local CHANNEL orders_out
    MODE NO_ACK RETRY POLICY BACKOFF 250ms MAX 30s
    ENCODE USING order_codec
  INHERIT ALL
  FLUSH EACH 100ms MAX BATCH SIZE 1MiB
  ON MESSAGE ERROR LOG
  ON GENERAL ERROR LOG;

COMMIT;

START;

Send An Order

Subscribe to the output channel in one terminal:

docker exec redis redis-cli SUBSCRIBE orders_out

Produce an order in another:

docker exec -i broker /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 --topic orders

{"order_id":"o-1001","customer":"acme","status":"new","amount":1500,"quantity":3}

The Redis subscriber prints the record back as JSON. You can also watch any relay from inside Nervix with a read-only session subscription (Sessions):

nervix-cli --domain quickstart subscribe watch orders

The pipeline works end to end. Next, add logic to it in Conditional Routing.