【1】引入第三方Bahir提供的Flink-redis相关依赖包
<!-- 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】Flink连接Redis并输出Sink处理结果
package com.zzx.flink
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}
object RedisSinkTest {
def main(args: Array[String]): Unit = {
// 创建一个流处理执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
//从文件中读取数据并转换为 类
val inputStreamFromFile: DataStream[String] = env.readTextFile("E:\\Project\\flink\\src\\main\\resources\\wordcount.txt")
//转换 SensorReading为用户自定义的类，是从文件转换而来的
val dataStream: DataStream[SensorReading] = inputStreamFromFile
.map( data => {
var dataArray = data.split(",")
SensorReading(dataArray(0),dataArray(1).toLong,dataArray(2).toDouble)
})
//定义一个 redis 的配置类 继承了FlinkJedisConfigBase 正是 SensorReading需要传入的参数，底层将有些数据保存成了状态数据。
val conf = new FlinkJedisPoolConfig.Builder().setHost("192.168.52.131").setPort(6379).setPassword("zzx").build()
//定义 RedisMapper 数据保存的类型
val myMapper = new RedisMapper[SensorReading] {
//定义保存数据到 redis的命令，hset table key value
override def getCommandDescription: RedisCommandDescription = {
// hset tablesname
new RedisCommandDescription(RedisCommand.HSET , "sensor_temp")
}
//设置key
override def getKeyFromData(data: SensorReading): String = data.id
//设置value
override def getValueFromData(data: SensorReading): String = data.temperature.toString
}
dataStream.addSink(new RedisSink[SensorReading](conf,myMapper))
env.execute("Redis Sink test")
}
}
查看源码可知RedisSink是继承自RichSinkFunction<IN>类文章来源：https://www.toymoban.com/news/detail-763703.html
public class RedisSink<IN> extends RichSinkFunction<IN> {
【3】查看Redis输出信息
文章来源地址https://www.toymoban.com/news/detail-763703.html
到了这里，关于Flink 输出至 Redis的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
