operator-api

name: operator-api description: "Use for dora operator development questions. Triggers on: DoraOperator, on_event, register_operator, DoraOutputSender, DoraStatus, operator:, operators:, runtime node, shared runtime, lightweight, 操作符, 运算符, 轻量级节点" globs: ["**/*.rs"] source: "https://docs.rs/dora-operator-api/latest/dora_operator_api/"

Operator API (dora-operator-api)

Lightweight alternative to nodes that run in a shared runtime process

Overview

Operators are lightweight alternatives to nodes:

• Run in a shared runtime process (not separate processes)
• Lower overhead than full nodes
• Ideal for simple transformations
• Currently best supported in Rust

Rust Operator

Dependencies

[dependencies]
dora-operator-api = "0.4"

[lib]
crate-type = ["cdylib"]

Basic Usage

use dora_operator_api::{
    register_operator, DoraOperator, DoraOutputSender, DoraStatus, Event
};
use dora_operator_api::arrow::array::*;

#[derive(Default)]
struct MyOperator {
    counter: u32,
}

impl DoraOperator for MyOperator {
    fn on_event(
        &mut self,
        event: &Event,
        output_sender: &mut DoraOutputSender,
    ) -> Result<DoraStatus, String> {
        match event {
            Event::Input { id, data } => {
                self.counter += 1;

                // Process input...

                // Send output
                let result = UInt32Array::from(vec![self.counter]);
                output_sender.send("count".to_string(), result)?;

                Ok(DoraStatus::Continue)
            }
            Event::InputClosed { id } => {
                println!("Input {} closed", id);
                Ok(DoraStatus::Continue)
            }
            Event::Stop => Ok(DoraStatus::Stop),
            _ => Ok(DoraStatus::Continue),
        }
    }
}

register_operator!(MyOperator);

Event Types

pub enum Event<'a> {
    // Input received
    Input {
        id: &'a str,
        data: ArrowData,
    },

    // Input parsing failed
    InputParseError {
        id: &'a str,
        error: String,
    },

    // Input closed by sender
    InputClosed {
        id: &'a str,
    },

    // Stop signal
    Stop,
}

DoraStatus

pub enum DoraStatus {
    Continue,  // Keep running
    Stop,      // Stop the operator
}

Sending Outputs

impl DoraOutputSender<'_> {
    /// Send output with Arrow data
    pub fn send(&mut self, id: String, data: impl Array) -> Result<(), String>;
}

Examples

use dora_operator_api::arrow::array::*;

// Integer output
let data = Int32Array::from(vec![1, 2, 3]);
output_sender.send("numbers".to_string(), data)?;

// String output
let data = StringArray::from(vec!["hello", "world"]);
output_sender.send("text".to_string(), data)?;

// Binary output
let data = BinaryArray::from(vec![b"bytes".as_slice()]);
output_sender.send("binary".to_string(), data)?;

Python Operator

Basic Usage

# my_operator.py
class Operator:
    def __init__(self):
        self.counter = 0

    def on_event(
        self,
        dora_event,
        send_output,
    ):
        if dora_event["type"] == "INPUT":
            self.counter += 1

            # Process input
            value = dora_event["value"]

            # Send output
            send_output("count", [self.counter])

        return DoraStatus.CONTINUE

Dataflow Configuration

Single Operator

nodes:
  - id: processor
    operator:
      python: my_operator.py
      # Or for Rust:
      # shared-library: target/release/libmy_operator
      inputs:
        data: source/output
      outputs:
        - processed

Multiple Operators

nodes:
  - id: runtime-node
    operators:
      - id: op1
        python: op1.py
        inputs:
          data: source/output
        outputs:
          - result1

      - id: op2
        python: op2.py
        inputs:
          input: op1/result1
        outputs:
          - result2

Rust Shared Library

nodes:
  - id: rust-operator
    operator:
      shared-library: target/release/my_operator
      inputs:
        data: source/output
      outputs:
        - processed

Complete Example: Counter Operator

Rust Implementation

// src/lib.rs
use dora_operator_api::{
    register_operator, DoraOperator, DoraOutputSender, DoraStatus, Event
};
use dora_operator_api::arrow::array::UInt64Array;

#[derive(Default)]
struct Counter {
    count: u64,
}

impl DoraOperator for Counter {
    fn on_event(
        &mut self,
        event: &Event,
        output_sender: &mut DoraOutputSender,
    ) -> Result<DoraStatus, String> {
        if let Event::Input { .. } = event {
            self.count += 1;

            let output = UInt64Array::from(vec![self.count]);
            output_sender.send("count".to_string(), output)?;
        }

        if matches!(event, Event::Stop) {
            return Ok(DoraStatus::Stop);
        }

        Ok(DoraStatus::Continue)
    }
}

register_operator!(Counter);

Cargo.toml

[package]
name = "counter-operator"
version = "0.1.0"
edition = "2024"

[lib]
crate-type = ["cdylib"]

[dependencies]
dora-operator-api = "0.4"

Dataflow

nodes:
  - id: timer
    path: timer-source
    inputs:
      tick: dora/timer/millis/1000
    outputs:
      - tick

  - id: counter
    operator:
      shared-library: target/release/libcounter_operator
      inputs:
        tick: timer/tick
      outputs:
        - count

  - id: logger
    path: logger-sink
    inputs:
      count: counter/count

Operators vs Nodes

Best Practices

1. Keep operators simple - Complex logic belongs in nodes
2. Avoid blocking operations - Don't block the runtime
3. Handle errors gracefully - Return appropriate DoraStatus
4. Use shared state carefully - Operators share address space

Related Skills

• node-api-rust - Full Rust nodes
• node-api-python - Full Python nodes
• dataflow-config - YAML configuration