High performance Rust stream processing engine seamlessly integrates AI capabilities, providing powerful real-time data processing and intelligent analysis.
See the code
English | 中文
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.
# Clone the repository
git clone https://github.com/arkflow-rs/arkflow.git
cd arkflow
# Build the project
cargo build --release
# Run tests
cargo test
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"
./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): Read data from Kafka topicsmqtt): Subscribe to messages from MQTT topicshttp): Receive data via HTTPfile): Reading data from files (CSV, JSON, Parquet, Avro, Arrow) with cloud storage supportgenerate): Generate synthetic test datasql): Query data from SQL databases (MySQL, PostgreSQL, SQLite)nats): Subscribe to messages from NATS topics with JetStream supportpulsar): Subscribe to messages from Pulsar topicsredis): Subscribe to messages from Redis streams, lists, or pub/sub channelswebsocket): Subscribe to messages from WebSocket connectionsmodbus): Read data from Modbus devicesmemory): In-memory data source for testingmultiple_inputs): Combine multiple input streams into one pipelineExample:
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_to_arrow / arrow_to_json): JSON data processing and transformationsql): Process data using SQL queriesarrow_to_protobuf / protobuf_to_arrow): Protobuf encoding/decodingbatch): Process messages in batchesvrl): Process data using VRLpython): Run Python user-defined functions over the batchembedding): Batch-embed a text column via an OpenAI-compatible API and append vector columnsllm): Send each row to an OpenAI-compatible chat completions API with bounded concurrency and append completionsvector_search): Search a Qdrant collection for top-k nearest neighbors of each row's vector and append matches as JSONmilvus_search): Search a Milvus collection for top-k nearest neighbors of each row's vector with one batched requestpgvector_search): Search a PostgreSQL table with the pgvector extension for top-k nearest neighbors of each row's vectorExample:
pipeline:
thread_num: 4
processors:
- type: json_to_arrow
- type: sql
query: "SELECT * FROM flow WHERE value >= 10"
ArkFlow supports multiple output targets:
kafka): Write data to Kafka topicsmqtt): Publish messages to MQTT topicshttp): Send data via HTTPinfluxdb): Write time-series data to InfluxDB 2.xmongodb): Write documents to a MongoDB collectionnats): Publish messages to NATS topicspulsar): Publish messages to Pulsar topicsredis): Write to Redis streams, lists, or pub/sub channelssql): Write to SQL databases (MySQL, PostgreSQL) with batch inserts and UPSERTstdout): Output data to the consoledrop): Discard dataqdrant): Upsert vectors and payloads into a Qdrant collectionpgvector): Upsert vectors and JSON payloads into a PostgreSQL table with the pgvector extensionmilvus): Upsert vectors and JSON payloads into a Milvus collection via the REST v2 APIExample:
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): 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): 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 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): 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): Encode/decode Arrow RecordBatches as JSON byte payloadsprotobuf): Encode/decode Arrow RecordBatches using a Protobuf descriptordebezium_json): Decode Debezium CDC envelope JSON from Kafka change-event topicsschema_registry): Decode Confluent wire-format Protobuf and Avro messages by resolving schemas from a Schema RegistryTemporary components provide external lookup state that processors can query at runtime:
redis): Redis-backed temporary lookup store (single node or cluster) read through a codecstreams:
- 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"
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!
(top 24 of 28)
1,422 followers · starred Mar 2025
325 followers · starred Apr 2025
302 followers · starred Apr 2025
429 followers · starred Jun 2025
High performance Rust stream processing engine seamlessly integrates AI capabilities, providing powerful real-time data processing and intelligent analysis.
See the code
English | 中文
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.
# Clone the repository
git clone https://github.com/arkflow-rs/arkflow.git
cd arkflow
# Build the project
cargo build --release
# Run tests
cargo test
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"
./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): Read data from Kafka topicsmqtt): Subscribe to messages from MQTT topicshttp): Receive data via HTTPfile): Reading data from files (CSV, JSON, Parquet, Avro, Arrow) with cloud storage supportgenerate): Generate synthetic test datasql): Query data from SQL databases (MySQL, PostgreSQL, SQLite)nats): Subscribe to messages from NATS topics with JetStream supportpulsar): Subscribe to messages from Pulsar topicsredis): Subscribe to messages from Redis streams, lists, or pub/sub channelswebsocket): Subscribe to messages from WebSocket connectionsmodbus): Read data from Modbus devicesmemory): In-memory data source for testingmultiple_inputs): Combine multiple input streams into one pipelineExample:
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_to_arrow / arrow_to_json): JSON data processing and transformationsql): Process data using SQL queriesarrow_to_protobuf / protobuf_to_arrow): Protobuf encoding/decodingbatch): Process messages in batchesvrl): Process data using VRLpython): Run Python user-defined functions over the batchembedding): Batch-embed a text column via an OpenAI-compatible API and append vector columnsllm): Send each row to an OpenAI-compatible chat completions API with bounded concurrency and append completionsvector_search): Search a Qdrant collection for top-k nearest neighbors of each row's vector and append matches as JSONmilvus_search): Search a Milvus collection for top-k nearest neighbors of each row's vector with one batched requestpgvector_search): Search a PostgreSQL table with the pgvector extension for top-k nearest neighbors of each row's vectorExample:
pipeline:
thread_num: 4
processors:
- type: json_to_arrow
- type: sql
query: "SELECT * FROM flow WHERE value >= 10"
ArkFlow supports multiple output targets:
kafka): Write data to Kafka topicsmqtt): Publish messages to MQTT topicshttp): Send data via HTTPinfluxdb): Write time-series data to InfluxDB 2.xmongodb): Write documents to a MongoDB collectionnats): Publish messages to NATS topicspulsar): Publish messages to Pulsar topicsredis): Write to Redis streams, lists, or pub/sub channelssql): Write to SQL databases (MySQL, PostgreSQL) with batch inserts and UPSERTstdout): Output data to the consoledrop): Discard dataqdrant): Upsert vectors and payloads into a Qdrant collectionpgvector): Upsert vectors and JSON payloads into a PostgreSQL table with the pgvector extensionmilvus): Upsert vectors and JSON payloads into a Milvus collection via the REST v2 APIExample:
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): 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): 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 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): 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): Encode/decode Arrow RecordBatches as JSON byte payloadsprotobuf): Encode/decode Arrow RecordBatches using a Protobuf descriptordebezium_json): Decode Debezium CDC envelope JSON from Kafka change-event topicsschema_registry): Decode Confluent wire-format Protobuf and Avro messages by resolving schemas from a Schema RegistryTemporary components provide external lookup state that processors can query at runtime:
redis): Redis-backed temporary lookup store (single node or cluster) read through a codecstreams:
- 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"
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!
(top 24 of 28)
1,422 followers · starred Mar 2025
325 followers · starred Apr 2025
302 followers · starred Apr 2025
429 followers · starred Jun 2025