Add Redis Sink (#655)

Co-authored-by: Remy Gwaramadze <gwaramadze@users.noreply.github.com>
Co-authored-by: Tim Sawicki <136370015+tim-quix@users.noreply.github.com>

Author: 3 people

LICENSES/LICENSE.redis-py

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2022-2023, Redis, inc.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

conda/post-link.sh

@@ -4,4 +4,5 @@ $PREFIX/bin/pip install \
 'rocksdict>=0.3,<0.4' \
 'protobuf>=5.27.2,<6.0' \
 'influxdb3-python>=0.7,<1.0' \
-'pyiceberg[pyarrow,glue]>=0.7,<0.8'
+'pyiceberg[pyarrow,glue]>=0.7,<0.8' \
+'redis[hiredis]>=5.2.0,<6'

docs/build/build.py

@@ -129,6 +129,7 @@
             "quixstreams.sinks.community.pubsub",
             "quixstreams.sinks.community.postgresql",
             "quixstreams.sinks.community.kinesis",
+            "quixstreams.sinks.community.redis",
         ]
     },
     # Order: base, core, community

docs/connectors/sinks/redis-sink.md

@@ -0,0 +1,133 @@
+# Redis Sink
+
+!!! info
+
+    This is a **Community** connector. Test it before using in production.
+
+    To learn more about differences between Core and Community connectors, see the [Community and Core Connectors](../community-and-core.md) page.
+
+Redis is an in-memory database that persists on disk.
+
+Quix Streams provides a sink to write processed data to Redis.
+
+## How To Install
+
+The dependencies for this sink are not included to the default `quixstreams` package.
+
+To install them, run the following command:
+
+```commandline
+pip install quixstreams[redis]
+```
+
+## How To Use
+
+To sink data to Redis, you need to create an instance of `RedisSink` and pass
+it to the `StreamingDataFrame.sink()` method:
+
+```python
+import json
+
+from quixstreams import Application
+from quixstreams.sinks.community.redis import RedisSink
+
+app = Application(
+    broker_address="localhost:9092",
+    auto_offset_reset="earliest",
+    consumer_group="consumer-group",
+)
+
+topic = app.topic("topic-name")
+
+# Initialize a sink
+redis_sink = RedisSink(
+    host="<Redis host>",
+    port="<Redis port>",
+    db="<Redis db>",
+    value_serializer=json.dumps,
+    key_serializer=None,
+    password=None,
+    socket_timeout=30.0,
+)
+
+sdf = app.dataframe(topic)
+sdf.sink(redis_sink)
+
+if __name__ == '__main__':
+    app.run()
+```
+
+## How It Works
+
+`RedisSink` is a batching sink.  
+It batches processed records in memory per topic partition, and writes them to Redis
+when a checkpoint has been committed.
+
+### Data serialization
+
+By default, `RedisSink` serializes records values to JSON and uses Kafka message keys as
+Redis keys.
+
+If you want to use a different encoding or change what keys will be inserted to Redis,
+you may use `key_serializer` and `value_serializer` callbacks.
+
+**Example**:
+
+Use a combination of record's key and value to create a new Redis key, 
+and convert values using the MessagePack format instead of JSON.
+
+```python
+from quixstreams import Application
+from quixstreams.sinks.community.redis import RedisSink
+
+# Assuming "msgpack-python" is installed
+import msgpack
+
+app = Application(
+    broker_address="localhost:9092",
+    auto_offset_reset="earliest",
+    consumer_group="consumer-group",
+)
+
+topic = app.topic("topic-name")
+
+redis_sink = RedisSink(
+    host="<Redis host>",
+    port="<Redis port>",
+    db="<Redis db>",
+    # Serialize records' values using msgpack format before writing to Redis
+    value_serializer=msgpack.dumps,
+    # Combine a new Redis key from the record's key and value.
+    key_serializer=lambda key, value: f'{key}-{value}',
+)
+
+sdf = app.dataframe(topic)
+sdf.sink(redis_sink)
+
+if __name__ == '__main__':
+    app.run()
+```
+
+### Atomic Writes
+
+`RedisSink`
+uses the [Redis Transactions](https://redis.io/docs/latest/develop/interact/transactions/)
+feature to ensure that all updates are executed atomically.
+
+## Delivery Guarantees
+
+`RedisSink` provides at-least-once guarantees, and the same records may be written
+multiple times in case of errors during processing.
+
+## Configuration
+
+Main configuration parameters:
+
+- `host`: a Redis db host.
+- `port`: a Redis db port.
+- `db`: a Redis db number.
+- `value_serializer`: a callable to serialize records' values.
+- `key_serializer`: a callable to serialize records' keys.
+
+For the full list of expected parameters, see
+the [RedisSink API](../../api-reference/sinks.md#redissink) page

pyproject.toml

@@ -37,6 +37,7 @@ all = [
     "psycopg2-binary>=2.9.9,<3",
     "boto3>=1.35.65,<2.0",
     "boto3-stubs>=1.35.65,<2.0",
+    "redis[hiredis]>=5.2.0,<6"
 ]
 
 avro = ["fastavro>=1.8,<2.0"]
@@ -48,6 +49,7 @@ bigquery = ["google-cloud-bigquery>=3.26.0,<3.27"]
 pubsub = ["google-cloud-pubsub>=2.23.1,<3"]
 postgresql = ["psycopg2-binary>=2.9.9,<3"]
 kinesis = ["boto3>=1.35.65,<2.0", "boto3-stubs[kinesis]>=1.35.65,<2.0"]
+redis=["redis[hiredis]>=5.2.0,<6"]
 
 [tool.setuptools.packages.find]
 include = ["quixstreams*"]

quixstreams/sinks/community/redis.py

@@ -0,0 +1,79 @@
+import json
+import logging
+import time
+from typing import Any, Callable, Optional, Union
+
+try:
+    import redis
+except ImportError as exc:
+    raise ImportError(
+        f"Package {exc.name} is missing: "
+        'run "pip install quixstreams[redis]" to use RedisSink'
+    ) from exc
+
+from quixstreams.sinks import BatchingSink, SinkBatch
+
+__all__ = ("RedisSink",)
+
+logger = logging.getLogger(__name__)
+
+
+class RedisSink(BatchingSink):
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        db: int,
+        value_serializer: Callable[[Any], Union[bytes, str]] = json.dumps,
+        key_serializer: Optional[Callable[[Any, Any], Union[bytes, str]]] = None,
+        password: Optional[str] = None,
+        socket_timeout: float = 30.0,
+        **kwargs,
+    ) -> None:
+        """
+        A connector to sink processed data to Redis.
+        It batches the processed records in memory per topic partition, and flushes them to Redis at the checkpoint.
+
+        :param host: Redis host.
+        :param port: Redis port.
+        :param db: Redis DB number.
+        :param value_serializer: a callable to serialize the value to string or bytes
+            (defaults to json.dumps).
+        :param key_serializer: an optional callable to serialize the key to string or bytes.
+            If not provided, the Kafka message key will be used as is.
+        :param password: Redis password, optional.
+        :param socket_timeout: Redis socket timeout.
+            Default - 30s.
+        :param kwargs: Additional keyword arguments passed to the `redis.Redis` instance.
+        """
+
+        super().__init__()
+        self._redis_uri = f"{host}:{port}/{db}"
+        self._client = redis.Redis(
+            host=host,
+            port=port,
+            db=db,
+            password=password,
+            socket_timeout=socket_timeout,
+            **kwargs,
+        )
+        self._key_serializer = key_serializer
+        self._value_serializer = value_serializer
+
+    def write(self, batch: SinkBatch) -> None:
+        # Execute Redis updates atomically using a transaction pipeline
+        start = time.monotonic()
+        with self._client.pipeline(transaction=True) as pipe:
+            for item in batch:
+                key = item.key
+                if self._key_serializer is not None:
+                    key = self._key_serializer(key, item.value)
+                value = self._value_serializer(item.value)
+                pipe.set(key, value)
+            keys_updated = len(pipe)
+            pipe.execute(raise_on_error=True)
+        time_elapsed = round(time.monotonic() - start, 4)
+        logger.debug(
+            f'Updated {keys_updated} keys in a Redis database "{self._redis_uri}" '
+            f"time_elapsed={time_elapsed}s"
+        )

tests/requirements.txt

@@ -6,3 +6,4 @@ fastavro>=1.8,<2.0
 protobuf>=5.27.2
 influxdb3-python>=0.7.0,<1.0
 pyiceberg[pyarrow,glue]>=0.7,<0.8
+redis[hiredis]>=5.2.0,<6