说明

flink本身没有提供对 redis的支持，需要自定义实现 sink

另一种方式，apache 的 bahir 提供了 connector，可以将数据发送给 redis，但版本落后，scala只支持2.11，本教程中flink中scala是2.12，开发环境可用，官网详见 链接

案例

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

redis hash数据类型详见 链接

pom.xml

需要添加下面依赖：

<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.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.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 S6ToRedis {
    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服务器信息
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
                .setHost("localhost")
                .setPort(6379).build();
        sumSOSO.addSink(new RedisSink<>(conf,new SinkRedisMapper()));

        env.execute();
    }


    static class SinkRedisMapper implements RedisMapper<Tuple2<String,Integer>> {
        //指定redis命令，这里使用hash数据类型的hset命令，key为 flink
        @Override
        public RedisCommandDescription getCommandDescription() {

            return new RedisCommandDescription(RedisCommand.HSET,"flink");
        }
        // 指定key
        @Override
        public String getKeyFromData(Tuple2<String, Integer> data) {
            return data.f0;
        }
        // 指定value
        @Override
        public String getValueFromData(Tuple2<String, Integer> stringIntegerTuple2) {
            return stringIntegerTuple2.f1.toString();
        }
    }

}

启动windows的redis

双击 redis-server.exe 运行

测试

运行 flink 程序

执行结果

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

原文出处：/show_1IX45UCVE9bb.html