Dataflow YAML Specification

Dataflows are defined in YAML files. Each file describes a graph of nodes, their inputs/outputs, and execution parameters.

A JSON Schema is available at the repo root (dora-schema.json) for editor autocompletion and validation.

Quick Start

nodes:
  - id: sender
    path: sender.py
    outputs:
      - message

  - id: receiver
    path: receiver.py
    inputs:
      message: sender/message

Run with dora run dataflow.yml (local mode) or dora up && dora start dataflow.yml (networked mode).

Editor Setup

Add a schema comment at the top of your YAML file for VS Code autocompletion (requires the YAML extension):

# yaml-language-server: $schema=https://raw.githubusercontent.com/dora-rs/dora/main/dora-schema.json
nodes:
  - id: my-node
    # ... autocompletion works here

Root-Level Fields

Field	Type	Default	Description
`nodes`	list	required	List of node configurations
`strict_types`	bool	`false`	Treat type warnings as errors in `validate` and `build`
`type_rules`	list	`[]`	User-defined type compatibility rules (see Type Annotations)
`health_check_interval`	float	`5.0`	Seconds between daemon health check sweeps. For each node with `health_check_timeout` set, the daemon checks whether the node has communicated within its timeout; if not, the node is killed and its `restart_policy` is evaluated
`exit_when_nodes_finish`	bool	`false`	Finish the dataflow once every node has, treating `dora/timer/...` inputs as a clock rather than as work. A timer input has no upstream node, so it never closes: by default a node consuming one is never told its inputs are done and the graph cannot end on its own. Overridden by `--exit-when-nodes-finish[=BOOL]` on `dora run` and `dora start` (see Completion)
`_unstable_deploy`	object	--	Root-level deployment config (see Deployment)
`_unstable_debug`	object	--	Debug options (see Debug)

Completion

By default a dataflow ends when every node has exited. A node is told its inputs are closed only when all of them are, and a dora/timer/... input never closes -- it has no upstream node that could finish it. So a graph in which any node consumes a timer cannot end on its own, even after every node doing real work has exited:

nodes:
  - id: worker
    path: ./worker
    inputs:
      data: producer/out
      tick: dora/timer/millis/100   # never closes

Set exit_when_nodes_finish to make a node finish once its data inputs have closed, with the timer treated as a clock rather than as work:

exit_when_nodes_finish: true

Off by default, and usually only wanted for batch-style runs: for a long-lived dataflow the timer is precisely what keeps it alive, and such a dataflow is normally ended with dora stop.

Nodes with no data inputs at all -- timer-only sources, or nodes with no inputs -- are unaffected. They have no dependency that could finish, so they are treated as sources and are never told to stop.

The command line overrides this field in either direction:

dora run flow.yml --exit-when-nodes-finish          # force on
dora start flow.yml --exit-when-nodes-finish=false  # force off
dora start flow.yml                                 # the YAML decides

Node Configuration

Every node requires an id. All other fields are optional (though most nodes need at least path or operator/operators).

Identity

Field	Type	Description
`id`	string	Required. Unique identifier. Must not contain `/`. Whitespace is discouraged
`name`	string	Human-readable display name (metadata only, used in tooling and logs)
`description`	string	Documentation string (metadata only, not used at runtime)

Source

A node's executable comes from a local path, a git repository, a hub package, a module reference, or is implicit (operator/ROS2 nodes).

Field	Type	Description
`path`	string	Path to executable or script. Can also be a URL (legacy)
`module`	string	Path to a module definition file (mutually exclusive with `path`). See Modules Guide
`git`	string	Git repo URL. `dora build` clones it and uses the clone dir as working directory
`branch`	string	Branch to checkout (requires `git`, mutually exclusive with `tag`/`rev`)
`tag`	string	Tag to checkout (requires `git`, mutually exclusive with `branch`/`rev`)
`rev`	string	Commit hash to checkout (requires `git`, mutually exclusive with `branch`/`tag`)
`hub`	string	A Hub package reference `[<namespace>/]<name>@<version-req>` (mutually exclusive with `path`/`git`/`build`). `dora build` resolves it to a pinned commit (or binary) and injects its typed contracts. Unstable. See Hub guide
`build`	string	Build commands run during `dora build`. Each line runs separately. With `--uv`, `pip`/`pip3` lines use `uv pip` and Python nodes get a dedicated managed venv at `<working-dir>/.dora/python-envs/<node-id>/` that the runtime reuses at spawn time. See CLI reference for details.
`args`	string	Command-line arguments (space-separated)

Example with git source:

- id: rust-node
  git: https://github.com/dora-rs/dora.git
  branch: main
  build: cargo build -p example-node --release
  path: target/release/example-node

Example with a hub source (resolution provides the build + path):

- id: detector
  hub: dora-yolo@^0.5        # [namespace/]name@semver-req; bare name = dora-rs/
  inputs:
    image: camera/image
  outputs:
    - bbox

Data I/O

Inputs

Inputs subscribe to another node's output using the format <node-id>/<output-id>:

inputs:
  # Short form
  image: camera/frames
  tick: dora/timer/millis/100

  # Long form with options
  sensor_data:
    source: sensor/frames
    queue_size: 10
    queue_policy: drop_oldest
    input_timeout: 5.0

  # Lossless input (blocks sender when full)
  commands:
    source: controller/cmd
    queue_size: 100
    queue_policy: backpressure

Input option	Type	Default	Description
`source`	string	required	`<node-id>/<output-id>` or timer path
`queue_size`	integer	`10`	Input buffer size
`queue_policy`	string	`drop_oldest`	`drop_oldest`: drops oldest message when full. `backpressure`: buffers up to 10x `queue_size` without dropping (drops with ERROR log at hard cap)
`input_timeout`	float	--	Circuit breaker timeout in seconds. If no message arrives within this period, the daemon closes the input and the node receives an `InputClosed` event for graceful degradation

Built-in Timers

Timers are virtual nodes that emit ticks at fixed intervals:

inputs:
  tick: dora/timer/millis/100   # every 100ms
  slow: dora/timer/millis/1000  # every 1s
  fast: dora/timer/hz/30        # 30 Hz (~33ms)

Built-in Log Aggregation

Subscribe to structured log messages from all (or filtered) nodes:

inputs:
  all_logs: dora/logs               # all nodes, all levels
  errors:   dora/logs/error         # error+ from all nodes
  sensor:   dora/logs/info/sensor   # info+ from specific node

Each message arrives as a JSON-encoded LogMessage string. See Logging for details.

Outputs

A list of output identifiers the node produces:

outputs:
  - processed_image
  - metadata

Type Annotations

Optional type annotations for inputs and outputs. Types are never required -- unannotated ports remain fully dynamic.

- id: camera
  path: camera.py
  outputs:
    - image
    - depth
  output_types:
    image: std/media/v1/Image
    depth: std/media/v1/Image

- id: detector
  path: detect.py
  inputs:
    image: camera/image
  input_types:
    image: std/media/v1/Image
  outputs:
    - bbox
  output_types:
    bbox: std/vision/v1/BoundingBox

Field	Type	Default	Description
`output_types`	object	`{}`	Maps output IDs to type URNs. Keys must match entries in `outputs`
`input_types`	object	`{}`	Maps input IDs to expected type URNs. Keys must match entries in `inputs`
`output_metadata`	object	`{}`	Maps output IDs to lists of required metadata keys
`pattern`	string	--	Communication pattern shorthand: `service-server`, `service-client`, `action-server`, `action-client`

Type URNs use the format std/<category>/v<version>/<TypeName> and support parameters (e.g. std/media/v1/AudioFrame[sample_type=f32]). See the Type Annotations Guide for the full standard type library, parameterized types, compatibility rules, and user-defined types.

Run dora validate <file> to check type annotations statically. For runtime checking, set DORA_RUNTIME_TYPE_CHECK=warn or error:

dora validate dataflow.yml
DORA_RUNTIME_TYPE_CHECK=warn dora run dataflow.yml

Types also appear on dora graph edge labels when annotated.

Module Parameters

When using module:, pass configuration values via params::

- id: fast_pipeline
  module: modules/transform.module.yml
  inputs:
    data: sender/value
  params:
    speed: "2.0"
    mode: turbo

Inside the module, params are available as $PARAM_<UPPERCASE_KEY> in args: and as environment variables. See the Modules Guide for full documentation.

Environment

env:
  MY_VAR: "value"          # string
  DEBUG: true               # boolean
  PORT: 8080                # integer
  RATE: 1.5                 # float
  FROM_HOST:
    __dora_env: HOST_VAR   # read from host environment at runtime

Environment variables apply to both build commands and node execution. Values support $VAR expansion syntax.

Some names are reserved and are dropped (with a warning in the daemon log) when set on a node:

Reserved	Why
`DORA_NODE_CONFIG`, `DORA_RUNTIME_CONFIG`	The daemon's own handle to the node — dataflow id, node id, and how to reach the daemon
`DORA_ZENOH_LISTEN`, `DORA_ZENOH_CONNECT`, `DORA_ZENOH_MULTICAST`, `ZENOH_CONFIG`	Node-to-node wiring. Overriding these produces a dataflow that starts cleanly and then exchanges nothing. Set `ZENOH_CONFIG` in the daemon's own environment instead — nodes inherit it
`DORA_RUN_PARENT_PID`	Names the process whose death ends the node. `dora run` sets it so a hard-killed CLI cannot strand nodes; a descriptor-supplied value would be an arbitrary self-destruct trigger
`LD_PRELOAD`, `LD_AUDIT`, `LD_LIBRARY_PATH`, `DYLD_INSERT_LIBRARIES`, `DYLD_LIBRARY_PATH`	Loader hijacking
`DORA_AUTH_TOKEN`, `DORA_ALLOW_SHELL_NODES`	Daemon-level security settings

Names that are empty or contain =, whitespace, or NUL are rejected as well.

Logging

Field	Type	Default	Description
`send_stdout_as`	string	--	Route raw stdout/stderr lines as a data output. Each line is sent as a separate Arrow message
`send_logs_as`	string	--	Route structured log entries as a data output. Each entry is a JSON string with fields: `timestamp`, `level`, `node_id`, `message`, `target`, `fields`
`min_log_level`	string	--	Suppress logs below this level from file output, coordinator forwarding, and `send_logs_as`. Levels from most to least verbose: `stdout` (all output including raw stdout), `trace`, `debug`, `info`, `warn`, `error`
`max_log_size`	string	--	Rotate log file at this size (e.g. `"50MB"`, `"1GB"`)
`max_rotated_files`	integer	`5`	Number of rotated log files to keep

Example:

- id: sensor
  path: ./sensor
  min_log_level: info
  send_stdout_as: raw_output
  send_logs_as: log_entries
  max_log_size: "100MB"
  max_rotated_files: 3
  outputs:
    - data
    - raw_output
    - log_entries

When using send_stdout_as or send_logs_as, include the output name in the outputs list so downstream nodes can subscribe to it.

For a complete guide to all logging features, see Logging.

Fault Tolerance

Field	Type	Default	Description
`restart_policy`	string	`never`	`never`, `on-failure`, or `always`
`max_restarts`	integer	`0`	Max restart attempts. 0 = unlimited
`restart_delay`	float	--	Initial backoff in seconds. Doubles each attempt
`max_restart_delay`	float	--	Cap for exponential backoff
`restart_window`	float	--	Time window for counting restarts. The counter resets after this many seconds since the first restart in the current window. Enables "N restarts per M seconds" semantics with `max_restarts`
`health_check_timeout`	float	--	Once the node has connected (subscribed to events), if it then does not communicate with the daemon (send outputs, acknowledge ticks, etc.) for this many seconds, the daemon kills the process and evaluates the `restart_policy`. Covers post-connection liveness only -- it does not bound startup time, so a node that hangs before it ever subscribes is not killed

Restart policies:

never (default): no automatic restart
on-failure: restart only on non-zero exit code
always: restart on any exit, except when stopped by user or all inputs closed with success

Example with exponential backoff:

- id: sensor
  path: ./sensor
  restart_policy: on-failure
  max_restarts: 5
  restart_delay: 1.0         # 1s, 2s, 4s, 8s, 16s
  max_restart_delay: 30.0    # capped at 30s
  restart_window: 300.0      # 5 restarts per 5 minutes
  health_check_timeout: 30.0

Arrow IPC Framing

By default, outputs use a raw Arrow buffer layout (zero-copy, no schema overhead). For self-describing wire format with full schema metadata, enable Arrow IPC framing per output:

- id: sensor
  path: ./sensor
  outputs:
    - image
  output_framing:
    image: arrow-ipc

When arrow-ipc framing is set, data is serialized as Arrow IPC stream format (schema + record batches). Receivers automatically detect the framing mode via a _framing metadata key and decode accordingly. This is useful when consumers need schema introspection or when interoperating with external Arrow tools.

Values: raw (default) or arrow-ipc.

CPU Affinity

Pin a node's process to specific CPU cores (Linux only, ignored on other platforms):

- id: realtime-controller
  path: ./controller
  cpu_affinity: [0, 1]
  inputs:
    sensor: sensor/data

The daemon applies sched_setaffinity before exec. Core indices must be less than the system's CPU_SETSIZE (typically 1024). Out-of-range cores are skipped with a warning.

Deployment

Assign nodes to specific machines using _unstable_deploy:

- id: camera-driver
  _unstable_deploy:
    machine: robot-arm
  path: ./target/debug/camera
  outputs:
    - frames

- id: ml-inference
  _unstable_deploy:
    machine: gpu-server
    labels:
      gpu: "true"
    distribute: scp
  path: ./target/debug/inference
  inputs:
    frames: camera-driver/frames

Deploy field	Type	Default	Description
`machine`	string	--	Target machine/daemon ID. The coordinator routes the node to the daemon registered with this ID
`working_dir`	string	--	Working directory on the target machine
`labels`	object	--	Key-value labels for scheduling. The coordinator matches these against labels reported by each daemon at registration
`distribute`	string	`local`	How built binaries reach the target daemon: `local` -- each daemon builds from source independently; `scp` -- CLI pushes the built binary via SSH/SCP before spawn; `http` -- daemon pulls the binary from the coordinator's HTTP artifact store

When nodes are on different machines, communication automatically switches from shared memory to Zenoh pub/sub.

Operator Nodes

Operators run in-process inside a shared runtime (no separate process). Use operator for a single operator or operators for multiple.

Single Operator

The id field is optional for single operators (defaults to the node id):

- id: detector
  operator:
    python: detect.py
    build: pip install -r requirements.txt
    inputs:
      image: camera/frames
    outputs:
      - bbox

Multiple Operators

Each operator in operators requires a unique id:

- id: runtime-node
  operators:
    - id: preprocessor
      shared-library: ../../target/debug/libpreprocess
      inputs:
        raw: sensor/data
      outputs:
        - processed
    - id: analyzer
      shared-library: ../../target/debug/libanalyze
      inputs:
        data: runtime-node/preprocessor/processed
      outputs:
        - result

Operator Source Types

Field	Description
`python`	Python script path, or `{source: "script.py", conda_env: "myenv"}`
`shared-library`	Path to a shared library (`.so`/`.dylib`/`.dll`)

Operators also support inputs, outputs, build, send_stdout_as, send_logs_as, min_log_level, max_log_size, and max_rotated_files with the same semantics as node-level fields.

ROS2 Bridge

Declare a node as a ROS2 bridge to automatically convert between ROS2 DDS messages and Dora's Arrow format. No custom code needed.

Single Topic

- id: camera_bridge
  ros2:
    topic: /camera/image_raw
    message_type: sensor_msgs/Image
    direction: subscribe
  outputs:
    - image

Multiple Topics

- id: robot_bridge
  ros2:
    topics:
      - topic: /camera/image_raw
        message_type: sensor_msgs/Image
        direction: subscribe
        output: image
      - topic: /cmd_vel
        message_type: geometry_msgs/Twist
        direction: publish
        input: velocity
    qos:
      reliable: true
  inputs:
    velocity: planner/cmd_vel
  outputs:
    - image

Service Bridge

- id: add_service
  ros2:
    service: /add_two_ints
    service_type: example_interfaces/AddTwoInts
    role: server
  inputs:
    request: client_node/request
  outputs:
    - response

Action Bridge

- id: nav_action
  ros2:
    action: /navigate
    action_type: nav2_msgs/NavigateToPose
    role: client
  inputs:
    goal: planner/goal
  outputs:
    - feedback
    - result

QoS Configuration

QoS can be set at the bridge level (applies to all topics) or per-topic:

QoS field	Type	Default	Description
`reliable`	bool	`false`	Reliable vs best-effort transport
`durability`	string	`volatile`	`volatile` or `transient_local`
`liveliness`	string	`automatic`	`automatic`, `manual_by_participant`, `manual_by_topic`
`lease_duration`	float	infinity	Lease duration in seconds
`max_blocking_time`	float	--	Max blocking time for reliable transport
`keep_last`	integer	`1`	History depth (KeepLast policy)
`keep_all`	bool	`false`	Use KeepAll history instead of KeepLast

Other ROS2 Fields

Field	Type	Default	Description
`namespace`	string	`/`	ROS2 namespace
`node_name`	string	node `id`	ROS2 node name

Debug

_unstable_debug:
  enable_debug_inspection: true

Required for dora topic echo, dora topic hz, and dora topic info commands.

Communication Patterns

Dora supports four communication patterns built on top of the dataflow:

Topic (default): pub/sub dataflow
Service: request/reply via request_id metadata
Action: goal/feedback/result via goal_id/goal_status metadata, with cancellation support
Streaming: session/segment/chunk via session_id/segment_id/seq/fin/flush metadata, with queue flush for interruption

See Communication Patterns for details and examples.

Full Example

health_check_interval: 10.0

_unstable_debug:
  enable_debug_inspection: true

nodes:
  - id: webcam
    operator:
      python: webcam.py
      inputs:
        tick: dora/timer/millis/100
      outputs:
        - image

  - id: detector
    operator:
      python: detect.py
      build: pip install ultralytics
      inputs:
        image: webcam/image
      outputs:
        - bbox

  - id: plotter
    operator:
      python: plot.py
      inputs:
        image: webcam/image
        bbox: detector/bbox

  - id: logger
    path: ./logger
    inputs:
      bbox: detector/bbox
    send_stdout_as: logs
    min_log_level: info
    restart_policy: on-failure
    max_restarts: 3
    outputs:
      - logs

Navigation Menu

Uh oh!

Dataflow YAML Specification

Quick Start

Editor Setup

Root-Level Fields

Completion

Node Configuration

Identity

Source

Data I/O

Inputs

Built-in Timers

Built-in Log Aggregation

Outputs

Type Annotations

Module Parameters

Environment

Logging

Fault Tolerance

Arrow IPC Framing

CPU Affinity

Deployment

Operator Nodes

Single Operator

Multiple Operators

Operator Source Types

ROS2 Bridge

Single Topic

Multiple Topics

Service Bridge

Action Bridge

QoS Configuration

Other ROS2 Fields

Debug

Communication Patterns

Full Example

Uh oh!

FilesExpand file tree

yaml-spec.md

Latest commit

History

yaml-spec.md

File metadata and controls

Dataflow YAML Specification

Quick Start

Editor Setup

Root-Level Fields

Completion

Node Configuration

Identity

Source

Data I/O

Inputs

Built-in Timers

Built-in Log Aggregation

Outputs

Type Annotations

Module Parameters

Environment

Logging

Fault Tolerance

Arrow IPC Framing

CPU Affinity

Deployment

Operator Nodes

Single Operator

Multiple Operators

Operator Source Types

ROS2 Bridge

Single Topic

Multiple Topics

Service Bridge

Action Bridge

QoS Configuration

Other ROS2 Fields

Debug

Communication Patterns

Full Example