1.导入依赖
<!-- https://mvnrepository.com/artifact/org.apache.bahir/flink-connector-redis -->
<dependency>
<groupId>org.apache.bahir</groupId>
<artifactId>flink-connector-redis_2.11</artifactId>
<version>1.0</version>
</dependency>
2.自定义RedisMapper类
例:将Datream中所有User样例类对象的ID写进key为Online_Number的set中
class MyRedisMapper extends RedisMapper[User]{
//定义对于每条数据使用的redis命令与描述
override def getCommandDescription: RedisCommandDescription = {
new RedisCommandDescription(RedisCommand.SADD , "Online_Number")
}
//定义每条数据的key值
override def getKeyFromData(user: User): String = "Online_Number"
//定义每条数据的value值
override def getValueFromData(user: User): String = user.id.toString
}
3.创建redis配置对象
redis sink提供三种类型的config类,分别是FlinkJedisPoolConfig、FlinkJedisClusterConfig、FlinkJedisSentinelConfig。
FlinkJedisPoolConfig:单节点的redis实例。
FlinkJedisPoolConfig:集群模式的redis实例。
FlinkJedisSentinelConfig:带哨兵模式的redis实例。
以FlinkJedisPoolConfig为例:
val config: FlinkJedisPoolConfig = FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build()
4.创建RedisSink对象
创建RedisSink对象,参数为第三步创建的config与第二步定义的RedisMapper对象。
val redisSink = new RedisSink[User](config,new MyRedisMapper)
5.调用Dstream的addSink
dataStream.addSink(redisSink)