学习 Flink（十五）：Redis Connector

更新至 Flink 1.8 版本

Apache Bahir 项目提供了基于 Flink DataStream API 的 Redis Connector，其内部使用了 Java Redis 客户端 jedis 实现了 Redis Sink。

依赖

编辑 pom.xml 文件，添加依赖：

<dependency>  
  <groupId>org.apache.bahir</groupId>
  <artifactId>flink-connector-redis_2.11</artifactId>
  <version>1.1-SNAPSHOT</version>
</dependency>

配置

Redis Sink 支持多种 Redis 环境：

• 单节点
• 集群
• 哨兵

以单节点连接池配置为例：

FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()  
        .setHost(ip)
        .setPassword(password)
        .setMaxTotal(maxTotal)
        .setMaxIdle(maxIdle)
        .setMinIdle(minIdle)
        build()

Redis Mapper

Redis Mapper 用于将输入数据映射为 Redis 操作，需要实现 RedisMapper<T> 接口。

RedisMapper 接口定义：

public interface RedisMapper<T> extends Function, Serializable {

    RedisCommandDescription getCommandDescription(); // ①

    String getKeyFromData(T data); // ②

    String getValueFromData(T data); // ③
}

① Redis 命令描述。当前支持的 Redis 命令和数据类型对应，如下表所示：

所有的 Redis 命令定义在 RedisCommand 枚举类型中，所有的 Redis 数据类型定义在 RedisDataType 枚举类型中。

注意：数据类型为 RedisDataType.HASH 和 RedisDataType.SORTED_SET ，需要提供额外的键信息。

SET 命令，对应数据类型 STRING：

new RedisCommandDescription(RedisCommand.SET)

HSET 命令，对应数据类型 HASH，需要额外键：

new RedisCommandDescription(RedisCommand.HSET, key)

② 返回键。如果数据类型为 RedisDataType.HASH 和 RedisDataType.SORTED_SET ，这里返回的键值是数据类型的键值。

③ 返回值。

Redis Sink

有了配置信息和 Redis Mapper，就可以创建 Redis Sink 写入数据到 Redis 了：

stream.addSink(new RedisSink<>(  
    config,
    mapper
));

参考

• Apache Flink Streaming Connector for Redis
• xetorthio/jedis: A blazingly small and sane redis java client
• Command reference - Redis