Flink Connectors怎么連接Redis

這篇文章主要介紹“Flink Connectors怎么連接Redis”，在日常操作中，相信很多人在Flink Connectors怎么連接Redis問(wèn)題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”Flink Connectors怎么連接Redis”的疑惑有所幫助！接下來(lái)，請(qǐng)跟著小編一起來(lái)學(xué)習(xí)吧！

通過(guò)使用Flink DataStream Connectors 數(shù)據(jù)流連接器連接到Redis緩存數(shù)據(jù)庫(kù)，并提供數(shù)據(jù)流輸入與輸出操作；

示例環(huán)境

java.version: 1.8.xflink.version: 1.11.1redis：3.2

示例數(shù)據(jù)源 （項(xiàng)目碼云下載）

Flink 系例 之 搭建開(kāi)發(fā)環(huán)境與數(shù)據(jù)

示例模塊 （pom.xml）

Flink 系例 之 DataStream Connectors 與 示例模塊

數(shù)據(jù)流輸入

DataStreamSource.java

package com.flink.examples.redis;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Protocol;

/**
 * @Description 從redis中讀取數(shù)據(jù)輸出到DataStream數(shù)據(jù)流中
 */
public class DataStreamSource {
    /**
     * 官方文檔：https://bahir.apache.org/docs/flink/current/flink-streaming-redis/
     */

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String key = "props";
        //實(shí)現(xiàn)RichSourceFunction抽象方法，加載數(shù)據(jù)源數(shù)據(jù)到流中
        DataStream<Tuple2<String, String>> dataStream = env.addSource(new RichSourceFunction<Tuple2<String, String>>(){
            private JedisPool jedisPool = null;
            @Override
            public void run(SourceContext<Tuple2<String, String>> ctx) throws Exception {
                jedisPool = new JedisPool(new JedisPoolConfig(), "127.0.0.1", 6379, Protocol.DEFAULT_TIMEOUT);
                Jedis jedis = jedisPool.getResource();
                try{
                    ctx.collect(Tuple2.of(key, jedis.get(key)));
                }catch(Exception e){
                    e.printStackTrace();
                }finally{
                    if (jedis != null){
                        //用完即關(guān)，內(nèi)部會(huì)做判斷，如果存在數(shù)據(jù)源與池，則回滾到池中
                        jedis.close();
                    }
                }
            }
            @Override
            public void cancel() {
                try {
                    super.close();
                }catch(Exception e){
                }
                if (jedisPool != null){
                    jedisPool.close();
                    jedisPool = null;
                }
            }
        });
        dataStream.print();
        env.execute("flink redis source");
    }

}

數(shù)據(jù)流輸出

DataStreamSink.java

package com.flink.examples.redis;

import org.apache.commons.lang3.RandomUtils;
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.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;

/**
 * @Description 將數(shù)據(jù)流寫入到redis中
 */
public class DataStreamSink {

    /**
     * 官方文檔：https://bahir.apache.org/docs/flink/current/flink-streaming-redis/
     */

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //1.寫入數(shù)據(jù)到流中
        String [] words = new String[]{"props","student","build","name","execute"};
        DataStream<Tuple2<String, Integer>> sourceStream = env.fromElements(words).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String v) throws Exception {
                return Tuple2.of(v, RandomUtils.nextInt(1000, 9999));
            }
        });
        sourceStream.print();

        //2.實(shí)例化FlinkJedisPoolConfig 配置redis
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").setPort(6379).build();

        //3.寫入到redis，實(shí)例化RedisSink，并通過(guò)flink的addSink的方式將flink計(jì)算的結(jié)果插入到redis
        sourceStream.addSink(new RedisSink<>(conf, new RedisMapper<Tuple2<String, Integer>>(){
            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.SET, null);
                //通過(guò)實(shí)例化傳參，設(shè)置hash值的key
                //return new RedisCommandDescription(RedisCommand.HSET, key);
            }
            @Override
            public String getKeyFromData(Tuple2<String, Integer> tuple2) {
                return tuple2.f0;
            }
            @Override
            public String getValueFromData(Tuple2<String, Integer> tuple2) {
                return tuple2.f1.toString();
            }
        }));
        env.execute("flink redis sink");
    }

}

數(shù)據(jù)展示

到此，關(guān)于“Flink Connectors怎么連接Redis”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章！