NebuLi: Distributed Deployment

NebulaStream provides two primary frontend interfaces for distributed deployments:

nes-repl - Distributed query controller for multi-node deployments via interactive REPL
nes-cli - Stateless one-shot CLI for deploying and controlling queries from topology files

Both interfaces support JSON output for programmatic access.

`nes-repl` (Interactive REPL)

Starting the REPL

nes-repl -d -f JSON

Flags:

-d - Debug mode with detailed logging
-f <format> - Output format: JSON for programmatic access, TEXT for tabular format (default: TEXT)
-s <address> - Server address to connect to (default: localhost:8080)
--on-exit <behavior> - Behavior when REPL exits (default: DO_NOTHING)
- DO_NOTHING - Exit immediately, leaving queries running on workers
- WAIT_FOR_QUERY_TERMINATION - Wait for all queries to finish before exiting
- STOP_QUERIES - Stop all running queries and wait for termination before exiting
-e <behavior> - Error handling behavior
- FAIL_FAST - Exit with non-zero code on first error (default for non-interactive mode)
- RECOVER - Ignore errors and continue (default for interactive mode)
- CONTINUE_AND_FAIL - Continue execution but return non-zero exit code at the end

Distributed Mode

Distributed mode requires explicit worker registration/removal before deploying queries. The HOST configuration is required for all sources and sinks to specify their target worker. Queries are always deployed based on the most recent topology state.

[!NOTE] nes-repl does not create or start workers - they must be started independently before registration.

[!WARNING] Removing workers from the topology will not terminate running queries. In general the behavior around changing topologies is not yet specified.

Single Worker Example

-- 1. Register a worker node
CREATE WORKER 'sink-node:8080' SET ('sink-node:9090' AS DATA);
-- Returns: [{"worker":"sink-node:8080"}]

-- 2. Create logical source
CREATE LOGICAL SOURCE endless(ts UINT64);

-- 3. Create physical source with host specification
CREATE PHYSICAL SOURCE FOR endless
TYPE Generator
SET(
    'ALL' as "SOURCE".STOP_GENERATOR_WHEN_SEQUENCE_FINISHES,
    'CSV' as INPUT_FORMATTER."TYPE",
    'emit_rate 10' AS "SOURCE".GENERATOR_RATE_CONFIG,
    10000000 AS "SOURCE".MAX_RUNTIME_MS,
    'sink-node:8080' AS "SOURCE"."HOST",  -- Specify target host (gRPC address)
    1 AS "SOURCE".SEED,
    'SEQUENCE UINT64 0 10000000 1' AS "SOURCE".GENERATOR_SCHEMA
);

-- 4. Create sink with host specification
CREATE SINK someSink(TS UINT64)
TYPE File
SET(
    'out.csv' as "SINK".FILE_PATH,
    'CSV' as "SINK".OUTPUT_FORMAT,
    'sink-node:8080' AS "SINK"."HOST"  -- Specify target host (gRPC address)
);

-- 5. Deploy query
SELECT TS FROM ENDLESS INTO SOMESINK;

Query status shows one global query status as well as potentially multiple local query statuses (one per worker).

Complex Multi-Node Example (8-Node Topology)

-- worker creation (multi-statement)
CREATE WORKER 'sink-node:8080' SET ('sink-node:9090' AS DATA);
CREATE WORKER 'source-node-1:8080' SET ('source-node-1:9090' AS DATA,
    'intermediate-node-1:8080' AS "DOWNSTREAM");
CREATE WORKER 'source-node-2:8080' SET ('source-node-2:9090' AS DATA,
    'intermediate-node-1:8080' AS "DOWNSTREAM");
CREATE WORKER 'source-node-3:8080' SET ('source-node-3:9090' AS DATA,
    'intermediate-node-2:8080' AS "DOWNSTREAM");
CREATE WORKER 'source-node-4:8080' SET ('source-node-4:9090' AS DATA,
    'intermediate-node-2:8080' AS "DOWNSTREAM");
CREATE WORKER 'source-node-5:8080' SET ('source-node-5:9090' AS DATA,
    'intermediate-node-2:8080' AS "DOWNSTREAM");
CREATE WORKER 'intermediate-node-1:8080' SET ('intermediate-node-1:9090' AS DATA,
    'sink-node:8080' AS "DOWNSTREAM");
CREATE WORKER 'intermediate-node-2:8080' SET ('intermediate-node-2:9090' AS DATA,
    'sink-node:8080' AS "DOWNSTREAM");

-- Deploy multiple queries to different nodes
SELECT ID, VALUE, TIMESTAMP
FROM Generator(..., 'source-node-1:8080' AS "SOURCE"."HOST", ...)
INTO Print('sink-node:8080' AS "SINK"."HOST", ...);

SELECT ID, VALUE, TIMESTAMP
FROM Generator(..., 'source-node-5:8080' AS "SOURCE"."HOST", ...)
INTO Print('sink-node:8080' AS "SINK"."HOST", ...);

-- Verify query distribution
SHOW QUERIES;
-- Returns: 8 total queries (2 global + 6 local instances across nodes)

-- Drop specific query
DROP QUERY WHERE ID='<query-id>';

`nes-cli` (One-Shot Topology Controller)

NES-CLI is a “stateless” CLI tool for deploying and managing queries based on YAML topology files. Unlike the REPL, nes-cli performs single operations and exits. The CLI is stateless in the sense that source/sink catalogs and topology configuration are always loaded from YAML files—nothing is persisted between invocations.

[!NOTE] Implementation Detail: To enable query management across CLI invocations, the CLI maintains an internal mapping of global query IDs to local query instances in $XDG_STATE_HOME/nebucli/ (or $HOME/.local/state/nebucli/). This is an implementation detail and should not be relied upon.

Basic Usage

# Display help
nes-cli --help

# Dump topology (validate and print parsed topology)
nes-cli -t topology.yaml dump
nes-cli -d -t topology.yaml dump  # With debug output

# Start query from topology file
nes-cli -t topology.yaml start

# Start ad-hoc query (override topology query)
nes-cli -t topology.yaml start 'SELECT * FROM GENERATOR_SOURCE INTO VOID_SINK'

# Check query status
nes-cli -t topology.yaml status <query-id>

# Stop query
nes-cli -t topology.yaml stop <query-id>

# Stop multiple queries
nes-cli -t topology.yaml stop <query-id-1> <query-id-2> <query-id-3>

# Use environment variable for topology file
export NES_TOPOLOGY_FILE=topology.yaml
nes-cli dump
nes-cli start

# Read topology from stdin
cat topology.yaml | nes-cli -t - dump
cat topology.yaml | nes-cli -t - start
cat topology.yaml | nes-cli -t - start 'SELECT * FROM GENERATOR_SOURCE INTO VOID_SINK'
cat topology.yaml | nes-cli -t - status <query-id>
cat topology.yaml | nes-cli -t - stop <query-id>

# Works with Docker too
cat topology.yaml | docker run -i nes-cli -t - start
cat topology.yaml | docker run -i nes-cli -t - dump
cat topology.yaml | docker run -i nes-cli -t - stop <query-id>
cat topology.yaml | docker run -i nes-cli -t - status <query-id>

Flags:

-t <file> - Topology file path, or - to read from stdin
-d - Debug mode with detailed logging

Topology File Resolution Order:

The CLI looks for the topology file in the following priority order:

-t <file> flag - Explicitly specified file path, or -t - to read from stdin
NES_TOPOLOGY_FILE environment variable
topology.yaml in current directory
topology.yml in current directory

Topology File Format

Topology files define the complete system state in YAML format, including workers, logical sources, physical sources, and sinks. The query field can contain 0, 1, or multiple query statements:

Omitted: No queries in topology file (use ad-hoc query via command line argument)
Single query: query: | SELECT ... (string)
Multiple queries: query: [...] (array of strings)

[!NOTE] Providing a query via the command line (e.g., nes-cli -t topology.yaml start 'SELECT ...') will override any queries defined in the topology file’s query field.

Example: Single Query Topology

query: |
  SELECT * FROM GENERATOR_SOURCE INTO VOID_SINK

sinks:
  - name: VOID_SINK
    host: worker-1:8080
    schema:
      - name: DOUBLE
        type: FLOAT64
    type: Void
    config: { }
    parser_config: { }

logical:
  - name: GENERATOR_SOURCE
    schema:
      - name: DOUBLE
        type: FLOAT64

physical:
  - logical: GENERATOR_SOURCE
    host: worker-1:8080
    parser_config:
      type: CSV
      field_delimiter: ","
    type: Generator
    source_config:
      generator_rate_type: FIXED
      generator_rate_config: emit_rate 10
      stop_generator_when_sequence_finishes: NONE
      seed: 1
      generator_schema: |
        NORMAL_DISTRIBUTION FLOAT64 0 1

workers:
  - host: worker-1:8080
    data_address: worker-1:9090
    max_operators: 10000

Example: Multi-Worker Topology with Data Routing

query: |
  SELECT * FROM GENERATOR_SOURCE INTO VOID_SINK

sinks:
  - name: VOID_SINK
    host: worker-2:8080 # sink located at worker-2
    schema:
      - name: DOUBLE
        type: FLOAT64
    type: Void
    config: { }
    parser_config: { }

logical:
  - name: GENERATOR_SOURCE
    schema:
      - name: DOUBLE
        type: FLOAT64

physical:
  - logical: GENERATOR_SOURCE
    host: worker-1:8080 # source located at worker-1
    parser_config:
      type: CSV
      field_delimiter: ","
    type: Generator
    source_config:
      generator_rate_type: FIXED
      generator_rate_config: emit_rate 1000
      stop_generator_when_sequence_finishes: NONE
      seed: 1
      generator_schema: |
        NORMAL_DISTRIBUTION FLOAT64 0 1

workers:
  - host: worker-1:8080
    data_address: worker-1:9090
    max_operators: 10000
    downstream: [ worker-2:8080 ]  # Route data to worker-2
  - host: worker-2:8080
    data_address: worker-2:9090
    max_operators: 10000

Model Registration

The topology file supports an optional models section for registering ML models. Models are registered before queries are submitted, so they can be referenced by MODEL_INFERENCE in the query.

models:
  - name: iris
    path: /path/to/iris.onnx
    input:
      - name: p1
        type: FLOAT32
      - name: p2
        type: FLOAT32
      - name: p3
        type: FLOAT32
      - name: p4
        type: FLOAT32
    output:
      - name: setosa
        type: FLOAT32
      - name: versicolor
        type: FLOAT32
      - name: virginica
        type: FLOAT32

Each model entry requires:

name - Identifier used in MODEL_INFERENCE(name, ...) queries
path - Absolute path to an .onnx model file (must exist at registration time)
input - List of input fields with name and type (must match the model’s input tensor)
output - List of output fields with name and type (must match the model’s output tensor)

For the equivalent SQL syntax and full usage examples, see the Query API guide.

Query Management

Checking Status:

nes-cli -t topology.yaml status <query-id>

Returns JSON array with query status information:

[
  {
    "query_id": "amazing_stallion",
    "query_status": "Running"
  },
  {
    "grpc_addr": "worker-1:8080",
    "query_status": "Running"
  },
  {
    "grpc_addr": "worker-2:8080",
    "query_status": "Running"
  }
]

Query Status Values:

"Running" - Query is actively processing
"PartiallyStopped" - Some query instances have stopped (e.g., source reached end of stream)
"Unreachable" - Cannot reach one or more workers. Affected LocalQueries will have the ConnectionError state.

Stopping Queries:

# Stop single query
nes-cli -t topology.yaml stop <query-id>

# Stop multiple queries
nes-cli -t topology.yaml stop <id1> <id2> <id3>