Flink教程-将流式数据写入redis
- 背景
- 实例讲解
- 引入pom
- 构造数据源
- 构造redis配置
- 实现RedisMapper
- 动态hash key
背景
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
完整的代码请参考:
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
- 【自然框架】QuickPager asp.net 分页控件的Ajax分页方式。
- 骑行在华盛顿 针对320万次共享单车骑行数据的分析
- 【自然框架】QuickPagerSQL——专门生成分页用的SQL的类库
- 【数据科学】数据科学中的 Spark 入门
- 【自然框架】QuickPager分页控件的单独的源码 V2.0.4.2。
- 用 iPhoneX 的 FaceID 刷脸解锁真的靠谱吗?
- 【自然框架】用CMS的栏目举例,聊一聊从“一层”到“三层”的变化
- java如何自定义鼠标指针
- 面向对象最重要的是“抽象”,三层最重要的也是“抽象”,没有抽象就不是真正的面向对象、三层。
- 唯一不变的就是一直在变”--“数据”的华丽“变身术”
- 借用 疯狂秀才 的页面,修改了一下自然框架后台管理的页面。
- PDF.NET(PWMIS数据开发框架)之SQL-MAP目标和规范
- PDF.NET数据开发框架 之SQL-MAP使用存储过程
- 用接口实现事件的一种方法,只是玩玩。
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- linux中alarm函数的实例讲解
- 如何给Linux虚拟机连上WiFi详解
- Linux系统查看CPU、机器型号、内存等信息
- 实战基本的Linux sed命令示例代码
- 在Linux中如何查看可用的网络接口详解
- Linux程序运行时加载动态库失败的解决方法
- Linux/CentOS系统同步网络时间的2种方法详解
- Linux 常用命令之Linux more命令使用方法
- Ubuntu下VIM配置成C++开发编辑器
- Ubuntu环境下使用G++编译CPP文件
- Linux下Tomcat的几种运行方式讲解
- linux中chmod命令用法详解
- Linux强制释放占用端口以及Linux防火墙端口开放方法详解
- CentOS7.5从零安装Python3.6.6的教程详解
- 实现一台或者多台Linux实例解绑SSH密钥对