Shulou(Shulou.com)06/01 Report--

This article mainly introduces "how to connect Flink Connectors to Redis". In daily operation, I believe many people have doubts about how to connect Flink Connectors to Redis. The editor consulted all kinds of materials and sorted out simple and easy-to-use methods of operation. I hope it will be helpful to answer the doubts about "how to connect Flink Connectors to Redis". Next, please follow the editor to study!

Connect to the Redis cache database by using the Flink DataStream Connectors data stream connector and provide data stream input and output operations

Sample environment

Java.version: 1.8.xflink.version: 1.11.1redis:3.2

Sample data source (project code cloud download)

Building Development Environment and data of Flink system example

Sample module (pom.xml)

DataStream Connectors and sample Module of Flink system

Data stream input

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 reads data from redis and outputs it to DataStream data stream * / public class DataStreamSource {/ * official document: 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" / implement the RichSourceFunction abstract method to load data source data into the stream 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) {/ / close immediately after use. Internal judgment will be made. If there is a data source and pool, it will be rolled back to the pool 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");}}

Data stream output

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 writes the data stream to redis * / public class DataStreamSink {/ * official document: https://bahir.apache.org/docs/flink/current/flink-streaming-redis/ * / public static void main (String [] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment (); / / 1. Write data to the stream 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. Instantiate the FlinkJedisPoolConfig configuration redis FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder (). SetHost ("127.0.0.1"). SetPort (6379). Build (); / / 3. Write to redis, instantiate RedisSink, and insert the result of flink calculation into redis sourceStream.addSink (new RedisSink (conf, new RedisMapper () {@ Override public RedisCommandDescription getCommandDescription () {return new RedisCommandDescription (RedisCommand.SET, null); / / set key / / return new RedisCommandDescription (RedisCommand.HSET, key) of hash value by instantiating parameters } @ Override public String getKeyFromData (Tuple2 tuple2) {return tuple2.f0;} @ Override public String getValueFromData (Tuple2 tuple2) {return tuple2.f1.toString ();}}); env.execute ("flink redis sink");}}

Data presentation

At this point, the study on "how to connect Flink Connectors to Redis" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.