add redis sink

Author: zhangblue

pom.xml

@@ -72,6 +72,9 @@
 
     <netty-all.version>4.1.50.Final</netty-all.version>
 
+
+    <bahir-redis.version>1.0</bahir-redis.version>
+
     <scope>compile</scope>
 
 
@@ -516,6 +519,14 @@
       <scope>${scope}</scope>
     </dependency>
 
+    <!--flink redis connector-->
+    <dependency>
+      <groupId>org.apache.bahir</groupId>
+      <artifactId>flink-connector-redis_${scala.version}</artifactId>
+      <version>${bahir-redis.version}</version>
+      <scope>${scope}</scope>
+    </dependency>
+
 
   </dependencies>
 

src/main/scala/com/zhangblue/sink/RedisSinkDemo.scala

@@ -0,0 +1,59 @@
+package com.zhangblue.sink
+
+import com.zhangblue.entity.TemperatureSensor
+import org.apache.flink.api.common.functions.MapFunction
+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}
+
+/**
+ * @author di.zhang
+ * @date 2020/7/14
+ * @time 23:52
+ **/
+object RedisSinkDemo {
+  def main(args: Array[String]): Unit = {
+
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+
+    env.setParallelism(1)
+
+    val inputStream: DataStream[String] = env.readTextFile("src/main/resources/temperature-sensor.txt")
+
+
+    val dataStream: DataStream[TemperatureSensor] = inputStream.map(new MyMapFunction())
+
+    //定义一个redis 的配置陪
+    val conf: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build()
+
+
+    dataStream.addSink(new RedisSink[TemperatureSensor](conf, new MyRedisMapper))
+
+    env.execute("redis sink demo")
+  }
+
+
+  private class MyMapFunction extends MapFunction[String, TemperatureSensor] {
+    override def map(value: String): TemperatureSensor = {
+      val splitData = value.split(",")
+      TemperatureSensor(splitData(0), splitData(1).toLong, splitData(2), splitData(3).toDouble, splitData(4))
+    }
+  }
+
+  private class MyRedisMapper extends RedisMapper[TemperatureSensor] {
+    //定义保存数据到redis的命令
+    override def getCommandDescription: RedisCommandDescription = {
+      new RedisCommandDescription(RedisCommand.HSET, "temperatureSensor_tmp")
+    }
+
+    override def getKeyFromData(data: TemperatureSensor): String = {
+      data.id
+    }
+
+    override def getValueFromData(data: TemperatureSensor): String = {
+      data.temperature.toString
+    }
+  }
+
+}

src/main/scala/com/zhangblue/wc/WordCount.scala

@@ -9,7 +9,7 @@ object WordCount {
     //创建一个批处理的执行环境
     val env = ExecutionEnvironment.getExecutionEnvironment
    //从文件中读取数据
-    val inputPath = "D:\\software\\workspace\\FlinkTutorial\\src\\main\\resources\\hello.txt"
+    val inputPath = "src/main/resources/hello.txt"
     val inputDataSet = env.readTextFile(inputPath)
 
     //分词之后做count