背景

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数据如下：

[图片上传失败...(image-c29e20-1597843550024)]

完整的代码请参考：

https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/connectors/redis/RedisSinkTest.java

动态生成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);
        }
    }

参考资料：
[1].https://github.com/apache/bahir-flink.git

更多内容，欢迎关注我的公众号【大数据技术与应用实战】

Flink教程-将流式数据写入redis