Flink怎么将流式数据写入redis

背景

redis作为一个高吞吐的存储系统，在生产中有着广泛的应用，今天我们主要讲一下如何将流式数据写入redis，以及遇到的一些问题 解决。官方并没有提供写入redis的connector，所以我们采用apache的另一个项目bahir-flink [1]中提供的连接器来实现。 

实例讲解 

引入pom

 <dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-connector-redis_2.11</artifactId>
   <version>1.1.5</version>
  </dependency>   
  

构造数据源

这里我们主要是模拟一条用户信息

  //user,subject,province
  Tuple3<String,String,String> tuple = Tuple3.of("tom", "math", "beijing");
  DataStream<Tuple3<String,String,String>> dataStream = bsEnv.fromElements(tuple);   
  

构造redis配置

• 单机配置

 FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("10.160.85.185")
                                                                // 可选 .setPassword("1234")
                                                                .setPort(6379)
                                                                .build();

• 集群配置

  InetSocketAddress host0 = new InetSocketAddress("host1", 6379);
  InetSocketAddress host1 = new InetSocketAddress("host2", 6379);
  InetSocketAddress host2 = new InetSocketAddress("host3", 6379);

  HashSet<InetSocketAddress> set = new HashSet<>();
  set.add(host0);
  set.add(host1);
  set.add(host2);

  FlinkJedisClusterConfig config = new FlinkJedisClusterConfig.Builder().setNodes(set)
                                                                        .build();   
  

实现RedisMapper

我们需要实现一个RedisMapper接口的类，这个类的主要功能就是将我们自己的输入数据映射到redis的对应的类型。

我们看下RedisMapper接口，这里面总共有三个方法：

• getCommandDescription：主要来获取我们写入哪种类型的数据，比如list、hash等等。
• getKeyFromData：主要是从我们的输入数据中抽取key
• getValueFromData：从我们的输入数据中抽取value

public interface RedisMapper<T> extends Function, Serializable {

 /**
  * Returns descriptor which defines data type.
  *
  * @return data type descriptor
  */
 RedisCommandDescription getCommandDescription();

 /**
  * Extracts key from data.
  *
  * @param data source data
  * @return key
  */
 String getKeyFromData(T data);

 /**
  * Extracts value from data.
  *
  * @param data source data
  * @return value
  */
 String getValueFromData(T data);
}

getCommandDescription方法返回一个RedisCommandDescription对象，我们看下RedisCommandDescription的构造方法：

 public RedisCommandDescription(RedisCommand redisCommand, String additionalKey) {
        ................
 }

 public RedisCommandDescription(RedisCommand redisCommand) {
  this(redisCommand, null);
 }

我们以数据写入hash结构为例，构造了一个key为HASH_NAME的RedisCommandDescription

new RedisCommandDescription(RedisCommand.HSET, "HASH_NAME");

两个构造方法区别就在于是否有第二个参数additionalKey，这个参数主要是针对SORTED_SET和HASH结构的，因为这两个结构需要有三个变量，其他的结构只需要两个变量就行了。

在hash结构里，这个additionalKey对应hash的key，getKeyFromData方法得到的数据对应hash的field，getValueFromData获取的数据对应hash的value。

最后我们数据写入对应的redis sink即可，写入的redis数据如下：

动态生成key

我们看到，上面我们构造redis的hash结构的时候，key是写死的，也就是只能写入一个key，如果我的key是动态生成的，该怎么办呢？

比如我有一个类似的需求，流式数据是一些学生成绩信息，我的key想要的是学生的name，field是相应的科目，而value是这个科目对应的成绩。

目前系统没提供这样的功能，不过这个也没事，没有什么不是改源码解决不了的。

我们看下RedisSink中的invoke方法，

 public void invoke(IN input) throws Exception {
  String key = redisSinkMapper.getKeyFromData(input);
  String value = redisSinkMapper.getValueFromData(input);

  switch (redisCommand) {
      ....................
   case HSET:
    this.redisCommandsContainer.hset(this.additionalKey, key, value);
    break;
   default:
    throw new IllegalArgumentException("Cannot process such data type: " + redisCommand);
  }
 }

我们看到对于hash结构来说，key和value也就是从我们的RedisMapper的实现类中获取的，但是additionalKey却不是动态生成的，我们只需要改下这里。动态获取additionalKey就行。

public interface RedisMapper<T> extends Function, Serializable{

 RedisCommandDescription getCommandDescription();

 String getKeyFromData(T data);

 String getValueFromData(T data);

 String getAdditionalKey(T data);
}

我们给RedisMapper接口添加一个getAdditionalKey方法，然后在实现类中实现该方法。

然后在RedisSink的invoke方法动态获取additionalKey,修改源码之后的方法如下：

 @Override
 public void invoke(IN input) throws Exception {
  String key = redisSinkMapper.getKeyFromData(input);
  String value = redisSinkMapper.getValueFromData(input);
  String additionalKey = redisSinkMapper.getAdditionalKey(input);
  switch (redisCommand) {
         ..................
   case HSET:
    this.redisCommandsContainer.hset(additionalKey, key, value);
    break;
   default:
    throw new IllegalArgumentException("Cannot process such data type: " + redisCommand);
  }
 }