Redis is an open source in-memory data structure storage system that can be used as a database, cache, and messaging middleware. It supports many types of data structures such as strings, hashes, lists, sets, ordered sets and range queries, bitmaps, hyperlogloglogs and geospatial index radius queries. Redis has built-in transactions and various levels of disk persistence, and provides high availability through Redis Sentinel and Cluster.

Apache Flink does not officially provide a connector for writing reids data. Apache StreamPark is based on Flink Connector Redis.

It encapsulates RedisSink, configures redis connection parameters, and automatically creates redis connections to simplify development. Currently, RedisSink supports the following connection methods: single-node mode, sentinel mode, and cluster mode because it does not support transactions.

StreamPark uses Redis’ MULTI command to open a transaction and the EXEC command to commit a transaction, see the link for details: http://www.redis.cn/topics/transactions.html, using RedisSink supports AT_LEAST_ONCE processing semantics by default. EXACTLY_ONCE semantics are supported with checkpoint enabled.

Redis is a key-value database, AT_LEAST_ONCE semantics flink job with abnormal restart the latest data will overwrite the previous version of data to achieve the final data consistency. If an external program reads the data during the restart, there is a risk of inconsistency with the final data.

EXACTLY_ONCE semantics will write to redis in batch when the flink job checkpoint is completed as a whole, and there will be a delay of checkpoint interval. Please choose the appropriate semantics according to the business scenario.

Redis Write Dependency

Flink Connector Redis officially provides two kinds, the following two api are the same, StreamPark is using org.apache.bahir dependency.

  1. <dependency>
  2.     <groupId>org.apache.bahir</groupId>
  3.     <artifactId>flink-connector-redis_2.11</artifactId>
  4.     <version>1.0</version>
  5. </dependency>
  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-connector-redis_2.10</artifactId>
  4.     <version>1.1.5</version>
  5. </dependency>

Writing Redis the Regular Way

The regular way of writing data using Flink Connector Redis is as follows:

1.Access to source

  1. import org.apache.flink.streaming.api.functions.source.SourceFunction;
  2. import scala.util.Random;
  4. public class TestSource implements SourceFunction<TestEntity> {
  6.     private boolean isRunning = true;
  8.     private Random random = new Random();
  10.     private int index = 0;
  12.     @Override
  13.     public void run(SourceContext<TestEntity> sourceContext) throws Exception {
  14.         while (isRunning && index <= 1000001) {
  15.             index += 1;
  16.             long userId = System.currentTimeMillis();
  17.             long orderId = random.nextInt(100);
  18.             int status = random.nextInt(1);
  19.             double price = random.nextDouble();
  20.             int quantity = new Random().nextInt(10);
  21.             TestEntity order = new TestEntity(userId, orderId, 1l, 1l, status, price, quantity, System.currentTimeMillis());
  22.             sourceContext.collect(order);
  23.         }
  24.     }
  26.     @Override
  27.     public void cancel() {
  28.         this.isRunning = false;
  29.     }
  30. }
  31. class TestEntity {
  32.     Long userId;
  33.     Long orderId;
  34.     Long siteId;
  35.     Long cityId;
  36.     Integer orderStatus;
  37.     Double price;
  38.     Integer quantity;
  39.     Long timestamp;
  41.     /**
  42.      * @param userId      : User ID
  43.      * @param orderId     : Order ID
  44.      * @param siteId      : Site ID
  45.      * @param cityId      : City ID
  46.      * @param orderStatus : Order status(1:Place order,0:Return order)
  47.      * @param price       : Unit price
  48.      * @param quantity    : Number of orders
  49.      * @param timestamp   : Order time
  50.      */
  51.     public TestEntity(Long userId, Long orderId, Long siteId, Long cityId, Integer orderStatus, Double price, Integer quantity, Long timestamp) {
  52.         this.userId = userId;
  53.         this.orderId = orderId;
  54.         this.siteId = siteId;
  55.         this.cityId = cityId;
  56.         this.orderStatus = orderStatus;
  57.         this.price = price;
  58.         this.quantity = quantity;
  59.         this.timestamp = timestamp;
  60.     }
  61. }

2. Write to redis

  1. import org.apache.flink.api.common.functions.MapFunction;
  2. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4. import org.apache.flink.streaming.connectors.redis.RedisSink;
  5. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
  6. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
  7. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
  8. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
  10. public class FlinkRedisSink {
  12.     public static void main(String[] args) throws Exception {
  13.         //1.Get the execution environment
  14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15.         env.setParallelism(1);
  17.         //2.Read data and convert to JavaBean
  18.         DataStreamSource<TestEntity> source = env.addSource(new TestSource(), TypeInformation.of(TestEntity.class));
  19.         //3.Write Data to Redis
  20.         FlinkJedisPoolConfig jedisPoolConfig = new FlinkJedisPoolConfig.Builder()
  21.                 .setHost("localhost")
  22.                 .setPort(6379)
  23.                 .build();
  24.         source.addSink(new RedisSink<>(jedisPoolConfig, new MyRedisMapper()));
  26.         //4.Perform the task
  27.         env.execute();
  28.     }
  30.     public static class MyRedisMapper implements RedisMapper<TestEntity> {
  31.         @Override
  32.         public RedisCommandDescription getCommandDescription() {
  33.             // Returns the type of data that exists in Redis, stored as a hash, with the second parameter being the outside key
  34.             return new RedisCommandDescription(RedisCommand.HSET, "");
  35.         }
  37.         @Override
  38.         public String getKeyFromData(TestEntity data) {
  39.             // Get Key from data: Key of Hash
  40.             return String.valueOf(data.userId());
  41.         }
  43.         @Override
  44.         public String getValueFromData(TestEntity data) {
  45.             // Get Value from data: Value of Hash
  46.             return String.valueOf(data.price());
  47.         }
  48.     }
  50. }

The above creation of FlinkJedisPoolConfig is tedious, and each operation of redis has to build RedisMapper, which is very insensitive. StreamPark uses a convention over configuration and automatic configuration. This only requires configuring redis StreamPark automatically assembles the source and sink parameters, which greatly simplifies the development logic and improves development efficiency and maintainability.

Apache StreamPark™ Writes to Redis

RedisSink defaults to AT_LEAST_ONCE (at least once) processing semantics, two-stage segment submission supports EXACTLY_ONCE semantics with checkpoint enabled, available connection types: single-node mode, sentinel mode.

1. Configure policy and connection information

Single-node configuration

  1. #redis sink configuration
  2. redis.sink:
  3.   host: 127.0.0.1 #Required parameters
  4.   #Optional parameters
  5.   port: 6379
  6.   database: 2
  7.   password:
  8.   connectType: jedisPool #Optional parameters: jedisPool（默认）|sentinel
  9.   maxTotal:
  10.   maxIdle:
  11.   minIdle:
  12.   connectionTimeout:

Sentinel mode configuration

  1. #redis sink configuration
  2. redis.sink:
  3.   masterName: master # Sentinel mode parameters
  4.   host: 192.168.0.1:6379, 192.168.0.3:6379 # Required parameter, must specify the port of the connection
  5.   connectType: sentinel
  6.   #Optional parameters
  7.   soTimeout: 6379
  8.   database: 2
  9.   password:
  10.   maxTotal:
  11.   maxIdle:
  12.   minIdle:
  13.   connectionTimeout:

2. Write to Redis

Writing to redis with StreamPark is very simple, the code is as follows:

Scala

  2. import org.apache.streampark.flink.core.scala.FlinkStreaming
  3. import org.apache.streampark.flink.core.scala.sink.{RedisMapper, RedisSink}
  4. import org.apache.flink.api.scala._
  5. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand
  6. import org.json4s.DefaultFormats
  8. object FlinkRedisSinkApp extends FlinkStreaming {
  10.   @transient
  11.   implicit lazy val formats: DefaultFormats.type = org.json4s.DefaultFormats
  13.   override def handle(): Unit = {
  15.     /**
  16.      * Create the source of read data
  17.      */
  18.     val source = context.addSource(new TestSource)
  21.     // Redis sink..................
  22.     //1)Define RedisSink
  23.     val sink: RedisSink = RedisSink()
  24.     //2)Write Mapper's mapping
  25.     val personMapper: RedisMapper[TestEntity] = RedisMapper[TestEntity](RedisCommand.HSET, "flink_user", _.userId.toString, _.orderId.toString)
  27.     sink.sink[TestEntity](source, personMapper, 60000000).setParallelism(1)
  29.   }
  31. }
  34. /**
  35.  * RedisMapper
  36.  * @param cmd redis -Write command
  37.  * @param additionalKey -Write additional keys, applicable to hset
  38.  * @param key -Write key
  39.  * @param value -Write value
  40.  * @tparam T
  41.  */
  42. case class RedisMapper[T](cmd: RedisCommand, additionalKey: String, key: T => String, value: T => String) extends RMapper[T] {
  44.   override def getCommandDescription: RedisCommandDescription = new RedisCommandDescription(cmd, additionalKey)
  46.   override def getKeyFromData(r: T): String = key(r)
  48.   override def getValueFromData(r: T): String = value(r)
  50. }

As the code shows, StreamPark automatically loads the configuration to create a RedisSink, and the user completes the redis write operation by creating the required RedisMapper object, additionalKey is the outermost key when hset is invalid for other write commands. RedisSink.sink() write the corresponding key corresponding to the data is required to specify the expiration time, if not specified default expiration time is java Integer.MAX_VALUE (67 years). As shown in the code.

  1. class RedisSink() extends Sink {
  3.   def sink[T](stream: DataStream[T], mapper: RedisMapper[T], ttl: Int = Int.MaxValue): DataStreamSink[T] = {
  4.     val sinkFun = (enableCheckpoint, cpMode) match {
  5.       case (false, CheckpointingMode.EXACTLY_ONCE) => throw new IllegalArgumentException("redis sink EXACTLY_ONCE must enable checkpoint")
  6.       case (true, CheckpointingMode.EXACTLY_ONCE) => new Redis2PCSinkFunction[T](config, mapper, ttl)
  7.       case _ => new RedisSinkFunction[T](config, mapper, ttl)
  8.     }
  9.     val sink = stream.addSink(sinkFun)
  10.     afterSink(sink, parallelism, name, uid)
  11.   }
  13. }

Supported redis operating commands

The following commands are supported for redis operations:

  1. public enum RedisCommand {
  3.     /**
  4.      * Insert the specified value at the head of the list stored at key.
  5.      * If key does not exist, it is created as empty list before performing the push operations.
  6.      */
  7.     LPUSH(RedisDataType.LIST),
  9.     /**
  10.      * Insert the specified value at the tail of the list stored at key.
  11.      * If key does not exist, it is created as empty list before performing the push operation.
  12.      */
  13.     RPUSH(RedisDataType.LIST),
  15.     /**
  16.      * Add the specified member to the set stored at key.
  17.      * Specified member that is already a member of this set is ignored.
  18.      */
  19.     SADD(RedisDataType.SET),
  21.     /**
  22.      * Set key to hold the string value. If key already holds a value,
  23.      * it is overwritten, regardless of its type.
  24.      */
  25.     SET(RedisDataType.STRING),
  27.     /**
  28.      * Adds the element to the HyperLogLog data structure stored at the variable name specified as first argument.
  29.      */
  30.     PFADD(RedisDataType.HYPER_LOG_LOG),
  32.     /**
  33.      * Posts a message to the given channel.
  34.      */
  35.     PUBLISH(RedisDataType.PUBSUB),
  37.     /**
  38.      * Adds the specified members with the specified score to the sorted set stored at key.
  39.      */
  40.     ZADD(RedisDataType.SORTED_SET),
  42.     /**
  43.      * Removes the specified members from the sorted set stored at key.
  44.      */
  45.     ZREM(RedisDataType.SORTED_SET),
  47.     /**
  48.      * Sets field in the hash stored at key to value. If key does not exist,
  49.      * a new key holding a hash is created. If field already exists in the hash, it is overwritten.
  50.      */
  51.     HSET(RedisDataType.HASH);
  52. }

RedisSink currently supports single-node mode and sentinel mode connections. And its cluster mode does not support transactions, but StreamPark is currently for support. Please call the official Flink Connector Redis api if you have a usage scenario.
Checkpoint must be enabled under EXACTLY_ONCE semantics, otherwise the program will throw parameter exceptions.
EXACTLY_ONCE semantics checkpoint data sink cache inside the memory, you need to reasonably set the checkpoint interval according to the actual data, otherwise there is a risk of oom.

Other Configuration

All other configurations must adhere to the StreamPark configuration, please refer to project configuration for specific configurable items and the role of each parameter.