Flink-06-DataStream API 之 sink--Redis实现

喜欢ヅ旅行 2022-02-25 10:48 207阅读 0赞

记录Flink从Socket获取数据，经过数据处理后将数据落地到Redis。

#### 添加依赖包 ####

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

##### 约定 #####

程序功能：接收Socket数据，将处理后的数据存到Redis。  
Redis数据类型：List  
命令：lpush  
key：l\_words  
value：Socket读入的数据  
测试：  
需要一台服务器（云主机/虚拟机）发送消息。

##### demo #####

package com.test.streaming.sink;
    
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    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:renxin.tang * @Desc: 功能：接收Socket数据，把数据保存到Redis中 * @Date: Created in 14:35 2019/4/1 */
    public class StreamingDemoToRedis { 
    
        public static void main(String[] args) throws Exception{ 
            // 获取流处理运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 获取数据源--cong socket 获取数据 主机名，端口
            DataStreamSource<String> text = env.socketTextStream("hwyuntrx", 9001, "\n");
            // 命令 key value
            // lpush l_words word
            // 对数据进行组装
            DataStream<Tuple2<String, String>> l_wordsData = text.map(new MapFunction<String, Tuple2<String, String>>() { 
                @Override
                public Tuple2<String, String> map(String value) throws Exception { 
                    return new Tuple2<String, String>("l_words", value);
                }
            });
            /** * 添加Redis sink 需要做三个事儿： * 1、创建自定义RedisMapper类，这个类需要实现RedisMapper，用来封装使用的Redis命令、获取key和获取值。在下面创建Redis sink的时候会用到 * 2、创建Redis配置，需要设置Redis主机和端口，你得让程序知道Redis在哪 * 3、创建Redis sink ，创建的时候需要传入上面的配置个实体类 */
            // 创建 Redis 配置
            FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("hwyuntrx").setPort(6379).build();
            // 创建Redis sink
            RedisSink<Tuple2<String, String>> redisSink = new RedisSink<>(conf, new MyRedisMapper());
            l_wordsData.addSink(redisSink);
            env.execute("StreamingDemoToRedis");
    
        }
        // 创建自定义RedisMapper类 这个地方注意Tuple2的包不要导入错 我用的Java实现，结果错导成了Scala包
        public static class MyRedisMapper implements RedisMapper<Tuple2<String,String>>{ 
            // 指定要使用的命令 -lpush 
            @Override
            public RedisCommandDescription getCommandDescription() { 
                return new RedisCommandDescription(RedisCommand.LPUSH);
            }
            // 表示从接收的数据中 获取需要操作的 Redis key
            @Override
            public String getKeyFromData(Tuple2<String, String> data) { 
                return data.f0;
            }
            // 表示从接收的数据中 获取需要操作的 Redis value
            @Override
            public String getValueFromData(Tuple2<String, String> data) { 
                return data.f1;
            }
        }
    }

### 测试 ###

1、开启Redis 服务  
      确保Redis能正常的连接，如果是新装的Redis要配置好参数，我一开始连接不上，启动程序的时候报错：

报错 DENIED Redis is running in protected mode because protected mode is enabled

百度原因后发现我的Redis是处在保护模式，仅本地能连接，这不完犊子了吗，给个链接，照着搞就行 [报错解决][Link 1]  
2、在云主机（虚拟机）上运行

nc -l 9001

3、idea启动Flink程序

4、在运行nc -l 9001的端口输入信息：  
![在这里插入图片描述][20190401175115738.jpg]  
5、打开Redis客户端查看是否写入

![在这里插入图片描述][20190401175229940.jpg]  
done

### 总结 ###

使用sink的时候都需要添加一个sink；

l_wordsData.addSink(redisSink);

添加Redis sink 需要做三个事儿：  
1、创建自定义RedisMapper类，这个类需要实现RedisMapper，用来封装使用的Redis命令、获取key和获取值。在下面创建Redis sink的时候会用到  
2、创建Redis配置，需要设置Redis主机和端口，你得让程序知道Redis在哪  
3、创建Redis sink ，创建的时候需要传入上面的配置个实体类

[Link 1]: https://blog.csdn.net/Agly_Clarlie/article/details/52251746
[20190401175115738.jpg]: /images/20220225/75dc7939abb44aa797b9dc155d476f40.png
[20190401175229940.jpg]: /images/20220225/35253f3b09a34d2485d5739f0f3247fd.png