object RedisSinkTest {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val streamFromFile = env.readTextFile("C:\\Users\\Mi\\Documents\\project\\idea\\FlinkTitorial\\src\\main\\resources\\sensor.txt")

    val dataStream: DataStream[SensorReading] = streamFromFile.map(d => {
      val arr = d.split(",")
      SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).toDouble)
    })

    //redis sink
    val config: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder().setHost("hadoop102").setPort(6379).build()
    dataStream.addSink(new RedisSink(config,new MyRedisMapper))

    env.execute("redis sink test")
  }

}

class MyRedisMapper extends RedisMapper[SensorReading]{

  //命令为hset，键为sensor_temperature
  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET,"sensor_temperature")
  }

  //field为传感器id
  override def getKeyFromData(t: SensorReading): String = t.id

  //value为温度
  override def getValueFromData(t: SensorReading): String = t.temperature.toString
}

127.0.0.1:6379> hgetall sensor_temperature
1) "sensor_1"
2) "35.80018327300259"
3) "sensor_6"
4) "15.402984393403084"
5) "sensor_10"
6) "38.101067604893444"
7) "sensor_7"
8) "6.720945201171228"

object EsSinkTest {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val streamFromFile = env.readTextFile("C:\\Users\\Mi\\Documents\\project\\idea\\FlinkTitorial\\src\\main\\resources\\sensor.txt")

    val dataStream: DataStream[SensorReading] = streamFromFile.map(d => {
      val arr = d.split(",")
      SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).toDouble)
    })

    //es sink
    val httpHosts = new util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("hadoop101",9200))
    //创建一个es sink的builder
    val esSinkBuilder = new ElasticsearchSink.Builder[SensorReading](httpHosts, new ElasticsearchSinkFunction[SensorReading] {
      override def process(t: SensorReading, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
        println("保存数据：" + t)
        //包装成map
        val map = new util.HashMap[String, String]()
        map.put("sensor_id", t.id)
        map.put("temperature", t.temperature.toString)
        map.put("ts", t.timestamp.toString)

        //创建index request，准备发送数据
        val indexRequest: IndexRequest = Requests.indexRequest().index("sensor").`type`("redingdata").source(map)

        //利用requestIndexer发送请求，写入数据
        requestIndexer.add(indexRequest)

        println("保存成功")
      }
    })
    esSinkBuilder

    dataStream.addSink(esSinkBuilder.build())

    env.execute("redis sink test")
  }

}