FlinkConnectors怎么连接Redis
这篇文章主要介绍“Flink Connectors怎么连接redis”,在日常操作中,相信很多人在Flink Connectors怎么连接Redis问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Flink Connectors怎么连接Redis”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
成都创新互联公司专注于洛川企业网站建设,响应式网站建设,商城网站建设。洛川网站建设公司,为洛川等地区提供建站服务。全流程按需定制网站,专业设计,全程项目跟踪,成都创新互联公司专业和态度为您提供的服务
通过使用Flink DataStream Connectors 数据流连接器连接到Redis缓存数据库,并提供数据流输入与输出操作;
示例环境
java.version: 1.8.xflink.version: 1.11.1redis:3.2
示例数据源 (项目码云下载)
Flink 系例 之 搭建开发环境与数据
示例模块 (pom.xml)
Flink 系例 之 DataStream Connectors 与 示例模块
数据流输入
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中读取数据输出到DataStream数据流中 */ 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"; //实现RichSourceFunction抽象方法,加载数据源数据到流中 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){ //用完即关,内部会做判断,如果存在数据源与池,则回滚到池中 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"); } }
数据流输出
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 将数据流写入到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.写入数据到流中 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计算的结果插入到redis sourceStream.addSink(new RedisSink<>(conf, new RedisMapper >(){ @Override public RedisCommandDescription getCommandDescription() { return new RedisCommandDescription(RedisCommand.SET, null); //通过实例化传参,设置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"); } }
数据展示
到此,关于“Flink Connectors怎么连接Redis”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!
本文题目:FlinkConnectors怎么连接Redis
本文网址:http://scyanting.com/article/pccoed.html