這篇文章主要介紹“Flink Connectors怎么連接redis”，在日常操作中，相信很多人在Flink Connectors怎么連接Redis問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Flink Connectors怎么連接Redis”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

成都創(chuàng)新互聯(lián)公司專注于洛川企業(yè)網(wǎng)站建設(shè),響應(yīng)式網(wǎng)站建設(shè),商城網(wǎng)站建設(shè)。洛川網(wǎng)站建設(shè)公司,為洛川等地區(qū)提供建站服務(wù)。全流程按需定制網(wǎng)站，專業(yè)設(shè)計，全程項目跟蹤，成都創(chuàng)新互聯(lián)公司專業(yè)和態(tài)度為您提供的服務(wù)

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

示例環(huán)境

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

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

Flink 系例 之 搭建開發(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";
        //實現(xiàn)RichSourceFunction抽象方法，加載數(shù)據(jù)源數(shù)據(jù)到流中
        DataStream> dataStream = env.addSource(new RichSourceFunction>(){
            private JedisPool jedisPool = null;
            @Override
            public void run(SourceContext> 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)部會做判斷，如果存在數(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> sourceStream = env.fromElements(words).map(new MapFunction>() {
            @Override
            public Tuple2 map(String v) throws Exception {
                return Tuple2.of(v, RandomUtils.nextInt(1000, 9999));
            }
        });
        sourceStream.print();

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

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

}

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

到此，關(guān)于“Flink Connectors怎么連接Redis”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
網(wǎng)站題目：FlinkConnectors怎么連接Redis
URL鏈接：http://weahome.cn/article/pccoed.html