Flink Sink之Redis_flink sink redis

5.7.2 Redis

flink-connector-redis

查询Flink连接器，最简单的就是查询关键字flink-connector-

这里将Redis当作sink的输出对象。

  1. pom依赖
1

<!--        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>
1
2
3
4
5
6
7

2. 编写代码

   package com.zch.apitest.sink;
   
   import com.zch.apitest.beans.SensorReading;
   import org.apache.flink.streaming.api.datastream.DataStream;
   import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
   import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
   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;
   import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
   import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
   
   /**
    * Author: zhaoHui
    * Date: 2022/01/22
    * Time: 16:32
    * Description:
    */
   public class SinkTest2_Redis {
       public static void main(String[] args) throws Exception{
   
           StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
           env.setParallelism(1);
   
           // 读取文件
           DataStream<String> inputStream = env.readTextFile("F:\\JAVA\\bigdata2107\\zch\\flink\\src\\main\\resources\\Sensor.txt");
   
           SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(lines -> {
               String[] split = lines.split(",");
               return new SensorReading(split[0], new Long(split[1]), new Double(split[2]));
           });
   
           // 定义jedis连接配置
           FlinkJedisPoolConfig flinkJedisPoolConfig = new FlinkJedisPoolConfig
                   .Builder()
                   .setHost("192.168.235.10")
                   .setPort(6379)
                   .build();
   
           dataStream.addSink(new RedisSink<>(flinkJedisPoolConfig,new MyRedisMapper()));
   
           env.execute();
   
       }
   
       // 自定义RedisMapper
       public static class MyRedisMapper implements RedisMapper<SensorReading>{
           // 定义保存数据到redis的命令，存成hash表，hset sensor_temp id temperature
           @Override
           public RedisCommandDescription getCommandDescription() {
               return new RedisCommandDescription(RedisCommand.HSET,"sensor_temp");
           }
   
           @Override
           public String getKeyFromData(SensorReading sensorReading) {
               return sensorReading.getId();
           }
   
           @Override
           public String getValueFromData(SensorReading sensorReading) {
               return sensorReading.getTemperature().toString();
           }
       }
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59
   60
   61
   62
   63
   64

   3. 启动redis服务（我这里是docker里的）

   4. 启动Flink程序

   5. 查看Redis里的数据

   因为最新数据覆盖前面的，所以最后redis里呈现的是最新的数据。

   localhost:6379>hgetall sensor_temp
   1) "sensor_1"
   2) "37.1"
   3) "sensor_6"
   4) "15.4"
   5) "sensor_7"
   6) "6.7"
   7) "sensor_10"
   8) "38.1"
   1
   2
   3
   4
   5
   6
   7
   8
   9