学习 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 命令 Redis 数据类型
HSET HASH
RPUSH,LPUSH LIST
SADD SET
PUBLISH PUBSUB
SET STRING
PFADD HYPER LOG LOG
ZADD SORTED SET
ZREM SORTED SET

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

注意:数据类型为 RedisDataType.HASHRedisDataType.SORTED_SET,需要提供额外的键信息。

SET 命令,对应数据类型 STRING:

new RedisCommandDescription(RedisCommand.SET)  

HSET 命令,对应数据类型 HASH,需要额外键:

new RedisCommandDescription(RedisCommand.HSET, key)  

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

③ 返回值。

Redis Sink

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

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

参考