📄️ File
文件输入(Input)通过 DataFusion 读取 JSON / CSV / Parquet / Avro / Arrow 文件。它支持本地路径与云对象存储(S3、GCS、Azure、HTTP、HDFS),并可选地对文件数据执行 SQL,或与 Ballista 分布式引擎集成。
📄️ Generate
生成输入(Input)以固定的时间间隔产生合成文本消息(Message),主要用于测试、演示和基准测试。每次读取返回 batch_size 份相同的 context 字符串。
📄️ HTTP
HTTP 输入(Input)以 Axum HTTP 服务器的形式运行,接受发送到 address+path 的 POST 请求。请求体(JSON)会被解码并转发到流处理流水线(Pipeline)。支持可选的 CORS 以及 Basic/Bearer 认证。bind 失败同步可见取消一次未完成的 read 不会丢失已解码的消息(参见输入投递语义)。
📄️ Kafka
Kafka 输入(Input)使用消费者组(consumer group)从一个或多个 Apache Kafka 主题(Topic)消费消息(Message)。只有在下游输出确认写入之后,偏移量(offset)才会推进(enable.auto.offset.store=false),从而在崩溃时保证至少一次(at-least-once)投递。
📄️ Memory
内存输入(Input)从内存队列读取消息(Message),该队列可以在配置中预先填充初始消息。主要用于测试和开发。
📄️ Modbus
Modbus 输入(Input)以固定的时间间隔轮询 Modbus TCP 设备,支持四种寄存器类型:coils(线圈)、discreteinputs(离散输入)、holdingregisters(保持寄存器)和 input_registers(输入寄存器)。
📄️ MQTT
MQTT 输入(Input)连接到 MQTT broker,订阅一个或多个主题(Topic),并接收实时消息。作为通道型输入,它遵循引擎的取消安全契约:取消一次未完成的 read 不会丢失已解码的消息(参见输入投递语义)。
📄️ Multiple Inputs
多输入将多个独立的输入(Input)组件合并为一条逻辑流(Stream)。所有子输入并发读取,消息(Message)按到达顺序进入同一条流水线(Pipeline)。每个子输入可以携带一个 name,该名称会写入 meta_source,以便下游阶段区分消息来源。
📄️ NATS
NATS 输入(Input)连接到 NATS 服务器,支持两种模式取消一次未完成的 read 不会丢失已解码的消息(参见输入投递语义)。
📄️ Pulsar
Pulsar 输入(Input)订阅 Apache Pulsar 主题(Topic),支持四种订阅(Subscription)类型——exclusive / shared / failover / key_shared——并可选支持 Token 或 OAuth2 认证。新建订阅从最早的消息开始消费(既有积压会被读取,与 Kafka 输入的默认语义一致);即使 broker 停止投递新消息,消息确认(ack)也能在有界时间内完成。作为通道型输入,它遵循引擎的取消安全契约:取消一次未完成的 read 不会丢失已解码的消息(参见输入投递语义)。
📄️ Redis
Redis 输入(Input)从 Redis 读取数据,同时支持单机(standalone)与集群(cluster)两种连接模式,并提供 Subscribe(频道/模式)与 List 两种消费模式。作为通道型输入,它遵循引擎的取消安全契约:取消一次未完成的 read 不会丢失已解码的消息(参见输入投递语义)。
📄️ SQL
SQL 输入(Input)通过 DataFusion 执行 select_sql 查询,从数据库(MySQL、PostgreSQL、SQLite、DuckDB)或文件格式读取数据。Ballista 为可选项,用于分布式查询。
📄️ WebSocket
WebSocket 输入(Input)以客户端身份连接远程 WebSocket 服务器,解码每条入站消息(Message)并转发到流水线(Pipeline)中。当前实现仅支持客户端模式。作为通道型输入,它遵循引擎的取消安全契约:取消一次未完成的 read 不会丢失已解码的消息(参见输入投递语义)。