Implementation:ArroyoSystems Arroyo Redis Connector

Knowledge Sources	ArroyoSystems_Arroyo
Domains	Streaming, Connectors
Last Updated	2026-02-08 08:00 GMT

Overview

RedisConnector implements the Arroyo Connector and LookupConnector traits for Redis, supporting sink operations to string, list, and hash data structures, as well as key-based lookup tables.

Description

The Redis connector supports both standalone and clustered Redis deployments via RedisClient which wraps either a standard Client or ClusterClient. Sink targets are defined by the Target enum: StringTable (SET with optional TTL and key column), ListTable (RPUSH/LPUSH with optional max length and append/prepend operation), and HashTable (HSET with key prefix and field column). Lookup tables use RedisLookup to perform MGET operations for key-based joins. The connector validates column configurations against the schema, ensuring key and field columns exist and are non-nullable TEXT types. Connection profiles support optional username and password with environment variable substitution via VarStr. The connector exposes a key metadata field for lookup table primary key mapping.

Usage

Use RedisConnector when you need to write streaming query results to Redis data structures or perform key-based lookups against Redis in join operations.

Code Reference

Source Location

• Repository: ArroyoSystems_Arroyo
• File: crates/arroyo-connectors/src/redis/mod.rs

Signature

pub struct RedisConnector {}

pub(crate) enum RedisClient {
    Standard(Client),
    Clustered(ClusterClient),
}

impl Connector for RedisConnector {
    type ProfileT = RedisConfig;
    type TableT = RedisTable;

    fn name(&self) -> &'static str;
    fn from_config(&self, id: Option<i64>, name: &str, config: RedisConfig,
        table: RedisTable, schema: Option<&ConnectionSchema>) -> anyhow::Result<Connection>;
    fn make_operator(&self, profile: RedisConfig, table: RedisTable,
        config: OperatorConfig) -> anyhow::Result<ConstructedOperator>;
    fn make_lookup(&self, profile: RedisConfig, table: RedisTable,
        config: OperatorConfig, schema: Arc<Schema>) -> anyhow::Result<Box<dyn LookupConnector + Send>>;
}

Import

use arroyo_connectors::redis::RedisConnector;

I/O Contract

Inputs

Name	Type	Required	Description
connection	RedisConfigConnection	Yes	Single address or list of cluster addresses
connector_type	TableType	Yes	Sink { target } or Lookup
target	Target	Yes (for sink)	StringTable, ListTable, or HashTable configuration
username	Option<VarStr>	No	Redis username
password	Option<VarStr>	No	Redis password

Outputs

Name	Type	Description
Connection	Connection	Configured Redis connection (Sink or Lookup type)
ConstructedOperator	ConstructedOperator	RedisSinkFunc operator or RedisLookup connector

Usage Examples

CREATE TABLE redis_sink (
    key TEXT,
    value TEXT
) WITH (
    connector = 'redis',
    address = 'redis://localhost:6379',
    type = 'sink',
    target = 'string',
    'target.key_prefix' = 'user:',
    'target.key_column' = 'key',
    format = 'json'
);

Related Pages

• Environment:ArroyoSystems_Arroyo_Rust_Runtime