English | 中文

Docs | 0.5 release docs

High performance Rust stream processing engine seamlessly integrates AI capabilities, providing powerful real-time data processing and intelligent analysis. It not only supports multiple input/output sources and processors, but also enables easy loading and execution of machine learning models, enabling streaming data and inference, anomaly detection, and complex event processing.

ArkFlow enlisted in the CNCF Cloud Native Landscape.

• High Performance: Built on Rust and Tokio async runtime, offering excellent performance and low latency
• Durable Delivery: At-least-once by default via per-stream WAL durability, with optional exactly-once for transactional sinks
• Multiple Data Sources: Support for Kafka, MQTT, HTTP, files, SQL databases, and many other input/output sources
• Powerful Processing: Built-in SQL queries, Python UDFs, JSON processing, Protobuf encoding/decoding, batch processing, and VRL
• Streaming Codecs: JSON and Protobuf codecs, plus Debezium CDC envelopes and Confluent Schema Registry wire-format
• Control Plane: An optional Hub and web console to observe, configure, and operate multiple ArkFlow compute nodes as a fleet
• Extensible: Modular design, easy to extend with new input, buffer, output, and processor components

# Clone the repository
git clone https://github.com/arkflow-rs/arkflow.git
cd arkflow

# Build the project
cargo build --release

# Run tests
cargo test

1. Create a configuration file config.yaml:

logging:
  level: info
streams:
  - input:
      type: "generate"
      context: '{ "timestamp": 1625000000000, "value": 10, "sensor": "temp_1" }'
      interval: 1s
      batch_size: 10

    pipeline:
      thread_num: 4
      processors:
        - type: "json_to_arrow"
        - type: "sql"
          query: "SELECT * FROM flow WHERE value >= 10"

    output:
      type: "stdout"
    error_output:
      type: "stdout"

2. Run ArkFlow:

./target/release/arkflow --config config.yaml

ArkFlow uses YAML format configuration files, supporting the following main configuration items:

logging:
  level: info  # Log level: debug, info, warn, error

streams: # Stream definition list
  - input:      # Input configuration
    # ...
    pipeline:   # Processing pipeline configuration
    # ...
    output:     # Output configuration
    # ...
    error_output: # Error output configuration
    # ...
    buffer:     # Buffer configuration
    # ... 

ArkFlow supports multiple input sources:

• Kafka (kafka): Read data from Kafka topics
• MQTT (mqtt): Subscribe to messages from MQTT topics
• HTTP (http): Receive data via HTTP
• File (file): Reading data from files (CSV, JSON, Parquet, Avro, Arrow) with cloud storage support
• Generate (generate): Generate synthetic test data
• SQL (sql): Query data from SQL databases (MySQL, PostgreSQL, SQLite)
• NATS (nats): Subscribe to messages from NATS topics with JetStream support
• Pulsar (pulsar): Subscribe to messages from Pulsar topics
• Redis (redis): Subscribe to messages from Redis streams, lists, or pub/sub channels
• WebSocket (websocket): Subscribe to messages from WebSocket connections
• Modbus (modbus): Read data from Modbus devices
• Memory (memory): In-memory data source for testing
• Multiple Inputs (multiple_inputs): Combine multiple input streams into one pipeline

Example:

input:
  type: kafka
  brokers:
    - localhost:9092
  topics:
    - test-topic
  consumer_group: test-group
  client_id: arkflow
  start_from_latest: true

ArkFlow provides multiple data processors:

• JSON (json_to_arrow / arrow_to_json): JSON data processing and transformation
• SQL (sql): Process data using SQL queries
• Protobuf (arrow_to_protobuf / protobuf_to_arrow): Protobuf encoding/decoding
• Batch Processing (batch): Process messages in batches
• VRL (vrl): Process data using VRL
• Python (python): Run Python user-defined functions over the batch

Example:

pipeline:
  thread_num: 4
  processors:
    - type: json_to_arrow
    - type: sql
      query: "SELECT * FROM flow WHERE value >= 10"

ArkFlow supports multiple output targets:

• Kafka (kafka): Write data to Kafka topics
• MQTT (mqtt): Publish messages to MQTT topics
• HTTP (http): Send data via HTTP
• InfluxDB (influxdb): Write time-series data to InfluxDB 2.x
• MongoDB (mongodb): Write documents to a MongoDB collection
• NATS (nats): Publish messages to NATS topics
• Pulsar (pulsar): Publish messages to Pulsar topics
• Redis (redis): Write to Redis streams, lists, or pub/sub channels
• SQL (sql): Write to SQL databases (MySQL, PostgreSQL) with batch inserts and UPSERT
• Standard Output (stdout): Output data to the console
• Drop (drop): Discard data

Example:

output:
  type: kafka
  brokers:
    - localhost:9092
  topic:
    type: value
    value:
      type: value
      value: test-topic
  client_id: arkflow-producer

The error_output accepts any of the output components listed above and receives messages that failed processing. The most common choices are Kafka, HTTP, and Standard Output for debugging.

Example:

error_output:
  type: kafka
  brokers:
    - localhost:9092
  topic:
    type: value
    value: error-topic
  client_id: error-arkflow-producer

ArkFlow provides buffer capabilities to handle backpressure and temporary storage of messages:

• Memory Buffer (memory): Memory buffer, for high-throughput scenarios and window aggregation. Compiled Streams treat it as a no-op pass-through; its capacity/timeout options are ignored.
• Session Window (session_window): The Session Window buffer component provides a session-based message grouping mechanism where messages are grouped based on activity gaps. It implements a session window that closes after a configurable period of inactivity. Compiled into a processing-time window operator.
• Sliding Window (deprecated) (sliding_window): Deprecated and unreachable in Stream configs — the stream compiler rejects the sliding window buffer. Use an event-time sliding window operator in a Job DAG instead.
• Tumbling Window (tumbling_window): The Tumbling Window buffer component provides a fixed-size, non-overlapping windowing mechanism for processing message batches. It implements a tumbling window algorithm with configurable interval settings. Compiled into a processing-time window operator.

Example:

buffer:
  type: memory

The legacy join buffer is also deprecated and unreachable in Stream configs — the stream compiler rejects it with a migration message pointing to Job DAG configuration; it is not a standalone buffer type.

Codecs attach to inputs and outputs through the codec configuration to encode and decode message payloads:

• JSON (json): Encode/decode Arrow RecordBatches as JSON byte payloads
• Protobuf (protobuf): Encode/decode Arrow RecordBatches using a Protobuf descriptor
• Debezium JSON (debezium_json): Decode Debezium CDC envelope JSON from Kafka change-event topics
• Schema Registry (schema_registry): Decode Confluent wire-format Protobuf and Avro messages by resolving schemas from a Schema Registry

Temporary components provide external lookup state that processors can query at runtime:

• Redis (redis): Redis-backed temporary lookup store (single node or cluster) read through a codec

streams:
  - input:
      type: kafka
      brokers:
        - localhost:9092
      topics:
        - test-topic
      consumer_group: test-group

    pipeline:
      thread_num: 4
      processors:
        - type: json_to_arrow
        - type: sql
          query: "SELECT * FROM flow WHERE value > 100"

    output:
      type: kafka
      brokers:
        - localhost:9092
      topic:
        type: value
        value: test-topic

streams:
  - input:
      type: "generate"
      context: '{ "timestamp": 1625000000000, "value": 10, "sensor": "temp_1" }'
      interval: 1ms
      batch_size: 10000

    pipeline:
      thread_num: 4
      processors:
        - type: "json_to_arrow"
        - type: "sql"
          query: "SELECT count(*) FROM flow WHERE value >= 10 group by sensor"

    output:
      type: "stdout"

• Conalog(Country: South Korea)

ArkFlow Plugin Examples

ArkFlow is licensed under the Apache License 2.0.

Discord: https://discord.gg/CwKhzb8pux

If you like or are using this project to learn or start your solution, please give it a star⭐. Thanks!