chenquan:
github: https://github.com/chenquan/arkflow
高性能 Rust 流处理引擎，提供强大的数据流处理能力，支持多种输入输出源和处理器。
特性
- 高性能：基于 Rust 和 Tokio 异步运行时构建，提供卓越的性能和低延迟
- 多种数据源：支持 Kafka 、MQTT 、HTTP 、文件等多种输入输出源
- 强大的处理能力：内置 SQL 查询、JSON 处理、Protobuf 编解码、批处理等多种处理器
- 可扩展：模块化设计，易于扩展新的输入、输出和处理器组件
安装
从源码构建
# 克隆仓库
git clone https://github.com/chenquan/arkflow.git
cd arkflow
# 构建项目
cargo build --release
# 运行测试
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"
- type: "arrow_to_json"
output:
type: "stdout"
- 运行 ArkFlow：
./target/release/arkflow --config config.yaml
配置说明
ArkFlow 使用 YAML 格式的配置文件，支持以下主要配置项：
顶级配置
logging:
level: info # 日志级别：debug, info, warn, error
streams: # 流定义列表
- input: # 输入配置
# ...
pipeline: # 处理管道配置
# ...
output: # 输出配置
# ...
输入组件
ArkFlow 支持多种输入源：
- Kafka：从 Kafka 主题读取数据
- MQTT：从 MQTT 主题订阅消息
- HTTP：通过 HTTP 接收数据
- 文件：从文件读取数据
- 生成器：生成测试数据
- SQL：从数据库查询数据
示例：
input:
type: kafka
brokers:
- localhost:9092
topics:
- test-topic
consumer_group: test-group
client_id: arkflow
start_from_latest: true
处理器
ArkFlow 提供多种数据处理器：
- JSON：JSON 数据处理和转换
- SQL：使用 SQL 查询处理数据
- Protobuf：Protobuf 编解码
- 批处理：将消息批量处理
示例：
pipeline:
thread_num: 4
processors:
- type: json_to_arrow
- type: sql
query: "SELECT * FROM flow WHERE value >= 10"
- type: arrow_to_json
输出组件
ArkFlow 支持多种输出目标：
- Kafka：将数据写入 Kafka 主题
- MQTT：将消息发布到 MQTT 主题
- HTTP：通过 HTTP 发送数据
- 文件：将数据写入文件
- 标准输出：将数据输出到控制台
示例：
output:
type: kafka
brokers:
- localhost:9092
topic: output-topic
client_id: arkflow-producer
示例
Kafka 到 Kafka 的数据处理
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"
- type: arrow_to_json
output:
type: kafka
brokers:
- localhost:9092
topic: processed-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"
- type: "arrow_to_json"
output:
type: "stdout"
Read More
