背景
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)]
完整的代码请参考：
动态生成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
更多内容，欢迎关注我的公众号【大数据技术与应用实战】
网友评论