Flink：将Flink中的数据下沉到Redis

﹏ヽ暗。殇╰゛Y 2024-03-23 12:57 55阅读 0赞

### 依赖 ###

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.7.2</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.7.2</version>
        <scope>provided</scope>
    </dependency>	
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-redis_2.11</artifactId>
        <version>1.1.5</version>
    </dependency>

### 发送 ###

> netcat下载安装：https://blog.csdn.net/qq\_38762390/article/details/115789281

nc -lk 7777

### 代码 ###

import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    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;
     
    public class StreamToRedis {
     
        public static void main(String[] args) throws Exception {
     
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
     
            String hostname = "192.168.25.128";
            int port = 7777;
            DataStreamSource<String> data = env.socketTextStream(hostname, port);
            SingleOutputStreamOperator<Tuple2<String, String>> m_word = data.map(new MapFunction<String, Tuple2<String, String>>() {
                public Tuple2<String, String> map(String value) throws Exception {
                    return new Tuple2<String, String>("m_word", value);
                }
            });
     
            FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("192.168.25.129").setPort(6379).build();
            // 参数一：配置信息
            // 参数二：下沉mapper映射器用于指明谁是key，谁是value
            RedisSink<Tuple2<String, String>> redisSink = new RedisSink<Tuple2<String, String>>(conf, new MyMapper());
            // 将结果保存到redis中
            m_word.addSink(redisSink);
            env.execute("to redis");
        }
     
        /**
         * RedisMapper<Tuple2<String,String>> 泛型为Tuple2<String,String>为将来传递过来的数据类型
         * 将数据其映射为key和value
         */
        public static class MyMapper implements RedisMapper<Tuple2<String,String>>{
     
            public RedisCommandDescription getCommandDescription() {
     
                // redis命令就是怎样插入key和value
                return new RedisCommandDescription(RedisCommand.LPUSH);
            }
     
            public String getKeyFromData(Tuple2<String, String> data) {
                // 指定key
                return data.f0;
            }
     
            public String getValueFromData(Tuple2<String, String> data) {
                // 指定value
                return data.f1;
            }
        }
    }