Flink Connector Redis

You can control how Flink data is mapped to Redis values using the value.data.structure parameter in your DDL.

value.data.structure = 'column' (Default)

In this mode, the key is determined by the order of fields in the DDL. There is no explicit mapping of a primary key to a Redis key; instead, fields are mapped to specific parts of the Redis command.

Example (SET):

-- username is the key, passport is the value
CREATE TABLE sink_redis(username VARCHAR, passport VARCHAR) WITH ('command'='set');

Example (HSET):

-- name is the map key, subject is the field, score is the value
CREATE TABLE sink_redis(name VARCHAR, subject VARCHAR, score VARCHAR) WITH ('command'='hset');

value.data.structure = 'row'

In this mode, the entire row is taken as the value, with fields separated by the \01 delimiter.

Example (SET):

CREATE TABLE sink_redis(username VARCHAR, passport VARCHAR) WITH ('command'='set');
-- Key: username, Value: username\01passport

Example (HSET):

CREATE TABLE sink_redis(name VARCHAR, subject VARCHAR, score VARCHAR) WITH ('command'='hset');
-- Key: name, Field: subject, Value: name\01subject\01score

create table sink_redis(username VARCHAR, passport VARCHAR)  with ('command'='set') 

create table sink_redis(name VARCHAR, subject VARCHAR, score VARCHAR)  with ('command'='hset') 

create table sink_redis(username VARCHAR, passport VARCHAR) with ('command'='set') 
#key: username , value: username\01passport.

create table sink_redis(name VARCHAR, subject VARCHAR, score VARCHAR) 
key: name, field:subject, value: name\01subject\01score.

By default, Redis stores values as single strings. To store and retrieve multiple fields for a single key, set 'value.data.structure'='row'. This stores the entire row in the Redis value, with fields separated by the \01 (SOH) character.

When reading back with 'command'='get', the connector will parse this delimited string back into the specified Flink table columns.

-- Write multiple fields as a single row in Redis
create table sink_redis(uid VARCHAR, score double, score2 double) 
with (
  'connector'='redis',
  'host'='10.11.69.176',
  'port'='6379',
  'redis-mode'='single',
  'password'='****',
  'command'='SET',
  'value.data.structure'='row'
);

insert into sink_redis select * from (values ('1', 10.3, 10.1));
-- Redis value becomes: "1\x0110.3\x0110.1"

-- Read the row back using 'get'
create table join_table 
with (
  'command'='get', 
  'value.data.structure'='row'
) like sink_redis;

-- Join as usual
select s.uid, j.score, j.score2
from source_table as s
join join_table for system_time as of s.proc_time as j on j.uid = s.uid;

To use the connector in your Flink project, you can either import the JAR file into your Flink lib directory or add it as a Maven dependency.

Maven Dependency

If you are using Maven, add the following dependency to your pom.xml. Note that if you do not include the jar-with-dependencies classifier, you must manually provide the Lettuce and netty-transport-native-epoll dependencies.

<dependency>
  <groupId>io.github.jeff-zou</groupId>
  <artifactId>flink-connector-redis</artifactId>
  <version>1.4.3</version>
  <!-- Uncomment the line below if Lettuce dependencies are not provided elsewhere -->
  <!-- <classifier>jar-with-dependencies</classifier> -->
</dependency>

Manual Installation

1. Run mvn package to generate the JAR.
2. Import the generated flink-connector-redis-1.4.3.jar into your Flink lib folder.
3. If the standard JAR is used, ensure Lettuce(6.2.1) and netty-transport-native-epoll(4.1.82.Final) are available in the classpath. Alternatively, use flink-connector-redis-1.4.3-jar-with-dependencies.jar to include all dependencies.

<dependency>
  <groupId>io.github.jeff-zou</groupId>
  <artifactId>flink-connector-redis</artifactId>
  <version>1.4.3</version>
  <!-- When the Lettuce netty-transport-native-epoll dependency is not imported separately -->
    <!--            <classifier>jar-with-dependencies</classifier>-->
</dependency>

The following parameters can be used in the WITH clause of your Flink SQL DDL to configure the Redis connector.

Connection Settings

Lookup (Dimension Table) Settings

Sink & TTL Settings

Debugging & Resource Limiting