案例

pom.xml

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

java

package sink;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
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.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

public class S7ToRedis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> dss = env.fromElements("lilei", "lili", "lilei", "lucy", "lili");

        SingleOutputStreamOperator<Tuple2<String, Integer>> mapSOSO = dss.map(item -> Tuple2.of(item, 1))
                .returns(Types.TUPLE(Types.STRING,Types.INT));

        SingleOutputStreamOperator<Tuple2<String, Integer>> sumSOSO = mapSOSO.keyBy(item -> item.f0).sum(1);

        //定义redis服务器信息
        sumSOSO.addSink(new SinkHashRedis("localhost",6379,"flink3"));

        env.execute();
    }


    static class SinkHashRedis extends RichSinkFunction<Tuple2<String, Integer>> {

        private Jedis jedis = null;
        private String host;
        private int port; // 默认端口6379
        private String key;
        public SinkHashRedis(String host, int port,String key) {
            this.host = host;
            this.port = port;
            this.key = key;
        }


        @Override
        public void open(Configuration parameters) throws Exception {
            jedis = new Jedis(host, port);
        }

        @Override
        public void invoke(Tuple2<String, Integer> value, Context context) {
            jedis.hset(key, value.f0, value.f1.toString());
        }

        @Override
        public void close() throws Exception {
            super.close();
            jedis.close();
        }

    }
}

启动windows的redis

双击 redis-server.exe 运行

测试

运行 flink 程序

执行结果

由于使用 hash 数据类型，所以 flink 运算后，保存数据时，会自动更新，而不会出现多条记录

原文出处：/show_1IX45ZKOgnCw.html