Flink Redis Sink 基本用法

落日映苍穹つ 2022-02-27 12:34 524阅读 0赞

### 文章目录 ###

*   *  简介
     *  依赖
     *  类
     *  使用
     *   *  RedisMapper
         *  RedisCommand
     *  Demo
     *  参考

## 简介 ##

flink提供了专门操作redis的Redis Sink

## 依赖 ##

<dependency>
        <groupId>org.apache.bahir</groupId>
        <artifactId>flink-connector-redis_2.11</artifactId>
        <version>1.0</version>
    </dependency>

## 类 ##

Redis Sink 提供用于向Redis发送数据的接口的类。接收器可以使用三种不同的方法与不同类型的Redis环境进行通信：

<table> 
 <thead> 
  <tr> 
   <th>类</th> 
   <th>场景</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>FlinkJedisPoolConfig</td> 
   <td>单Redis服务器</td> 
   <td>适用于本地、测试场景</td> 
  </tr> 
  <tr> 
   <td>FlinkJedisClusterConfig</td> 
   <td>Redis集群</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>FlinkJedisSentinelConfig</td> 
   <td>Redis哨兵</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

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

## 使用 ##

Redis Sink 核心类是 RedisMappe 是一个接口，使用时我们要编写自己的redis操作类实现这个接口中的三个方法

### RedisMapper ###

public interface RedisMapper<T> extends Function, Serializable { 
    
        /** * 设置使用的redis数据结构类型，和key的名词 * 通过RedisCommand设置数据结构类型 * Returns descriptor which defines data type. * * @return data type descriptor */
        RedisCommandDescription getCommandDescription();
    
        /** * 设置value中的键值对 key的值 * Extracts key from data. * * @param data source data * @return key */
        String getKeyFromData(T data);
    
        /** * 设置value中的键值对 value的值 * Extracts value from data. * * @param data source data * @return value */
        String getValueFromData(T data);
    }

### RedisCommand ###

使用RedisCommand设置数据结构类型时和redis结构对应关系。

<table> 
 <thead> 
  <tr> 
   <th align="left">Data Type</th> 
   <th align="left">Redis Command [Sink]</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">HASH</td> 
   <td align="left"><a href="http://redis.io/commands/hset" rel="nofollow">HSET</a></td> 
  </tr> 
  <tr> 
   <td align="left">LIST</td> 
   <td align="left"><a href="http://redis.io/commands/rpush" rel="nofollow">RPUSH</a>, <a href="http://redis.io/commands/lpush" rel="nofollow">LPUSH</a></td> 
  </tr> 
  <tr> 
   <td align="left">SET</td> 
   <td align="left"><a href="http://redis.io/commands/rpush" rel="nofollow">SADD</a></td> 
  </tr> 
  <tr> 
   <td align="left">PUBSUB</td> 
   <td align="left"><a href="http://redis.io/commands/publish" rel="nofollow">PUBLISH</a></td> 
  </tr> 
  <tr> 
   <td align="left">STRING</td> 
   <td align="left"><a href="http://redis.io/commands/set" rel="nofollow">SET</a></td> 
  </tr> 
  <tr> 
   <td align="left">HYPER_LOG_LOG</td> 
   <td align="left"><a href="http://redis.io/commands/pfadd" rel="nofollow">PFADD</a></td> 
  </tr> 
  <tr> 
   <td align="left">SORTED_SET</td> 
   <td align="left"><a href="http://redis.io/commands/zadd" rel="nofollow">ZADD</a></td> 
  </tr> 
  <tr> 
   <td align="left">SORTED_SET</td> 
   <td align="left"><a href="http://redis.io/commands/zrem" rel="nofollow">ZREM</a></td> 
  </tr> 
 </tbody> 
</table>

## Demo ##

package com.yljphp.demo.sink
    
    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}
    import org.apache.flink.util.StringUtils
    
    object RedisBasicApp extends App {
    
      val env = StreamExecutionEnvironment.getExecutionEnvironment
    
      env.setParallelism(1)
    
      val text: DataStream[String] = env.socketTextStream("localhost", 9000)
    
      val data: DataStream[(String, String)] = text.filter(!StringUtils.isNullOrWhitespaceOnly(_))
        .flatMap(_.split(","))
        .map(("ylj_test_1", _))
        .map(x=>{
          println(x)
          x
        })
    
    
      val config: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
        .setHost("localhost")
        .setPort(6379)
        .build()
    
    
      val redisSink = new RedisSink[(String,String)](config,new MyRedisMapper)
    
      data.addSink(redisSink)
    
      env.execute("sink_demo")
    }
    
    class MyRedisMapper extends RedisMapper[(String, String)] {
      override def getCommandDescription: RedisCommandDescription = {
        new RedisCommandDescription(RedisCommand.SADD, "ylj_test_2")
      }
    
      override def getKeyFromData(data: (String, String)): String = data._1
    
      override def getValueFromData(data: (String, String)): String = data._2
    }

![在这里插入图片描述][2019032717345566.png]

## 参考 ##

[Flink Redis Connector][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lsanBocA_size_16_color_FFFFFF_t_70]: /images/20220227/03fdbde2c85746a7abede53f2e2094b6.png
[2019032717345566.png]: /images/20220227/89e16dc6485c4be6b4b2a78bdd219aac.png
[Flink Redis Connector]: http://bahir.apache.org/docs/flink/current/flink-streaming-redis/