Flink Redis Sink 自定义

﹏ヽ暗。殇╰゛Y 2022-02-27 12:36 448阅读 0赞

### 文章目录 ###

*   *  背景
     *  依赖
     *  SinkFunction & RichSinkFunction
     *  Demo
     *  参考

## 背景 ##

Flink 默认的 RedisMapper 不能设置key的过期时间。

## 依赖 ##

<dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>2.9.0</version>
    </dependency>

## SinkFunction & RichSinkFunction ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 3]

## Demo ##

package com.yljphp.demo.sink
    
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.StringUtils
    import redis.clients.jedis.Jedis
    
    object RedisCustomSink extends App {
    
      val env = StreamExecutionEnvironment.getExecutionEnvironment
    
      env.setParallelism(1)
    
      val text: DataStream[String] = env.socketTextStream("localhost", 9000)
    
      val data = text.filter(!StringUtils.isNullOrWhitespaceOnly(_))
        .flatMap(_.split(","))
        .map(MySaveInfo("RedisCustomSink", _))
    
    
      data.addSink(new CustomSinkToRedis)
    
      env.execute("sink_demo")
    }
    
    class CustomSinkToRedis extends RichSinkFunction[MySaveInfo] {
    
      var redisCon: Jedis = _
    
      override def open(parameters: Configuration): Unit = {
        super.open(parameters)
        this.redisCon = new Jedis("localhost", 6379)
      }
    
      override def close(): Unit = {
        super.close()
        if (this.redisCon != null) {
          this.redisCon.close()
        }
      }
    
    
      override def invoke(value: MySaveInfo, context: SinkFunction.Context[_]): Unit = {
        //super.invoke(value, context)
        println(value)
        println("sadd之前"+this.redisCon.ttl(value.key))
        this.redisCon.sadd(value.key, value.value)
        println("sadd之后"+this.redisCon.ttl(value.key))
        if (this.redisCon.ttl(value.key) == -1) {
          this.redisCon.expire(value.key, 60 * 60)
        }
        println("设置过期之后/每次更新后"+this.redisCon.ttl(value.key))
      }
    
    }
    
    case class MySaveInfo(key: String, value: String)

socket窗口输入123,345,234,123

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 4]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 5]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 6]

## 参考 ##

[如何自定义 Data Sink][Data Sink]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70]: /images/20220227/a4838b50013c4e8ea98ab261131edae3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 1]: /images/20220227/c833769e8d394b1ba17ae345568f416d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 2]: /images/20220227/7603a4682c0f41e3b05b39829a60f5b2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 3]: /images/20220227/22583dfaf55f411ead52254607e12bcd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 4]: /images/20220227/341247161a1b451b8645e14fc4a2ec56.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 5]: /images/20220227/2a9a790a217b41d7816facbd4db03fb1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70 6]: /images/20220227/4ea187251e43440b87dd10b16979bf22.png
[Data Sink]: http://www.54tianzhisheng.cn/2018/10/31/flink-create-sink/