Flink写入Redis集群 重写flink-connector-redis包

愛在花開的季節  金牌会员 | 2022-9-3 10:32:12 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 894|帖子 894|积分 2682

起因：使用flink的时候难免和redis打交道，相信大家都使用过flink-connector-redis来处理，但是当我想要使用RedisSink写入集群时，发现居然不支持使用密码，于是有了这篇笔记。
 
事情的经过是这样的，我准备用Flink往Redis写入数据，我照常引入flink-connector-redis包
  1.         <dependency>
  2.             <groupId>org.apache.flink</groupId>
  3.             <artifactId>flink-connector-redis_2.11</artifactId>
  4.             <version>1.1.5</version>
  5.         </dependency>
复制代码
然后洋洋洒洒写下如下代码：
  1. package org.cube.flink
  3. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  4. import org.apache.flink.streaming.connectors.redis.RedisSink
  5. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
  6. import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}
  8. import java.net.InetSocketAddress
  9. import java.util.HashSet
  11. /**
  12. * @Author : Lawrence
  13. * @Date : 2022/7/24 23:11
  14. * @Description : Flink结果写入Redis集群
  15. * @Version : 1.0.0
  16. * @Modification_Record:
  17. * Version  Date       Remark
  18. * v1.0.0   2022/7/24  First Create
  19. */
  20. object RedisClusterSink {
  21.   def main(args: Array[String]): Unit = {
  22.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  23.     env.setParallelism(1)
  25.     // source
  26.     import org.apache.flink.api.scala._
  27.     val source = env.fromElements("1 hadoop","2 spark","3 flink", "4 hive", "5 redis", "6 hbase")
  29.     // process
  30.     val tupleValue = source.map(_.split(" ")).map(x => (x(0), x(1)))
  32.     // redis config
  33.     val builder = new FlinkJedisPoolConfig.Builder
  34.     builder.setHost("cube01").setPort(7001).setPassword("123456")
  35.     val redisConf: FlinkJedisPoolConfig = builder.build()
  37.     // sink
  38.     val redisSink = new RedisSink[(String, String)](redisConf, new MyRedisMapper())
  40.     tupleValue.addSink(redisSink)
  42.     env.execute("RedisClusterSink")
  43.   }
  44. }
  46. class MyRedisMapper extends RedisMapper[(String, String)] {
  47.   override def getCommandDescription: RedisCommandDescription = {
  48.     new RedisCommandDescription(RedisCommand.SET)
  49.   }
  51.   override def getKeyFromData(t: (String, String)): String = t._1
  53.   override def getValueFromData(t: (String, String)): String = t._2
  54. }
复制代码
然后兴高采烈地点击了运行，控制台却给了我一抹中国红，
其中最后一条是这样说的：
  1. Caused by: redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 9842 192.168.20.132:7003
复制代码
哦哦，是因为我的Redis是集群模式，
这并难不倒我，
我只需要把FlinkJedisPoolConfig改成FlinkJedisClusterConfig就万事大吉了。
  1.     // redis config
  2.     val builder = new FlinkJedisClusterConfig.Builder
  3.     val inetSocketAddress = new InetSocketAddress("cube01", 7001)
  4.     val nodeSet = new HashSet[InetSocketAddress]()
  5.     nodeSet.add(inetSocketAddress)
  6.     builder.setNodes(nodeSet).setPassword("123456")
  7.     val redisConf: FlinkJedisClusterConfig = builder.build()
复制代码
可是，这个类并没有setPassword方法，事实上它连"password"这个属性都没有。
这并难不倒我。
先不设密码总行了吧？
燃鹅并不行，控制台又给了我一抹中国红，
他是这样说的：
  1. Caused by: redis.clients.jedis.exceptions.JedisDataException: NOAUTH Authentication required.
复制代码
呵呵，这可难不倒我，
我的本能反应是，应该到Maven仓库中找到新版的flink-connector-redis包。
燃鹅，当我搜索之后发现，这已经是最新版了。

这也难不倒我。 
FlinkJedisPoolConfig不是可以设置密码吗？
FlinkJedisClusterConfig不是可以访问集群吗？
如果我把他们两个的代码整合一下呢？是不是就好了。
于是我本能地把"FlinkJedisClusterConfig"改写成了"MyFlinkJedisClusterConfig"类，增加了password属性和对应的get，set方法。
  1. package org.cube.flink;
  3. /**
  4. * @Author : Lawrence
  5. * @Date : 2022/7/25 21:14
  6. * @Description : 包含了password的FlinkJedisClusterConfig
  7. * @Version : 1.0.0
  8. * @Modification_Record:
  9. * Version  Date       Remark
  10. * v1.0.0   2022/7/25  First Create
  11. */
  12. import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
  13. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
  14. import org.apache.flink.util.Preconditions;
  15. import redis.clients.jedis.HostAndPort;
  16. import redis.clients.jedis.Protocol;
  18. import java.net.InetSocketAddress;
  19. import java.util.HashSet;
  20. import java.util.Iterator;
  21. import java.util.Set;
  23. public class MyFlinkJedisClusterConfig extends FlinkJedisConfigBase {
  24.     private static final long serialVersionUID = 1L;
  25.     private final Set<InetSocketAddress> nodes;
  26.     private final int maxRedirections;
  27.     private int soTimeout;
  28.     private String password;
  30.     private MyFlinkJedisClusterConfig(Set<InetSocketAddress> nodes, int connectionTimeout, int soTimeout,
  31.                                       int maxRedirections, int maxTotal, int maxIdle, int minIdle, String password) {
  32.         super(connectionTimeout, maxTotal, maxIdle, minIdle);
  33.         Preconditions.checkNotNull(nodes, "Node information should be presented");
  34.         Preconditions.checkArgument(!nodes.isEmpty(), "Redis cluster hosts should not be empty");
  35.         this.nodes = new HashSet(nodes);
  36.         this.soTimeout = soTimeout;
  37.         this.maxRedirections = maxRedirections;
  38.         this.password = password;
  39.     }
  41.     public Set<HostAndPort> getNodes() {
  42.         Set<HostAndPort> ret = new HashSet();
  43.         Iterator var2 = this.nodes.iterator();
  45.         while(var2.hasNext()) {
  46.             InetSocketAddress node = (InetSocketAddress)var2.next();
  47.             ret.add(new HostAndPort(node.getHostName(), node.getPort()));
  48.         }
  50.         return ret;
  51.     }
  52.     public int getMaxRedirections() {
  53.         return this.maxRedirections;
  54.     }
  55.     public int getSoTimeout() { return this.soTimeout; }
  56.     protected String getPassword() { return this.password; }
  58.     public String toString() {
  59.         return "JedisClusterConfig{nodes=" + this.nodes + ", timeout=" + this.connectionTimeout
  60.                 + ", maxRedirections=" + this.maxRedirections + ", maxTotal=" + this.maxTotal
  61.                 + ", maxIdle=" + this.maxIdle + ", minIdle=" + this.minIdle + '}';
  62.     }
  64.     public static class Builder {
  65.         private Set<InetSocketAddress> nodes;
  66.         private int timeout = Protocol.DEFAULT_TIMEOUT;
  67.         private int maxRedirections = 5;
  68.         //新增属性
  69.         private int soTimeout = Protocol.DEFAULT_TIMEOUT;
  70.         private int maxTotal = GenericObjectPoolConfig.DEFAULT_MAX_TOTAL;
  71.         private int maxIdle = GenericObjectPoolConfig.DEFAULT_MAX_IDLE;
  72.         private int minIdle = GenericObjectPoolConfig.DEFAULT_MIN_IDLE;
  73.         //增加的属性
  74.         private String password;
  76.         public Builder() {
  77.         }
  79.         public MyFlinkJedisClusterConfig.Builder setNodes(Set<InetSocketAddress> nodes) {
  80.             this.nodes = nodes;
  81.             return this;
  82.         }
  84.         public MyFlinkJedisClusterConfig.Builder setTimeout(int timeout) {
  85.             this.timeout = timeout;
  86.             return this;
  87.         }
  89.         public MyFlinkJedisClusterConfig.Builder setSoTimeout(int soTimeout) {
  90.             this.soTimeout = soTimeout;
  91.             return this;
  92.         }
  94.         public MyFlinkJedisClusterConfig.Builder setMaxRedirections(int maxRedirections) {
  95.             this.maxRedirections = maxRedirections;
  96.             return this;
  97.         }
  99.         public MyFlinkJedisClusterConfig.Builder setMaxTotal(int maxTotal) {
  100.             this.maxTotal = maxTotal;
  101.             return this;
  102.         }
  104.         public MyFlinkJedisClusterConfig.Builder setMaxIdle(int maxIdle) {
  105.             this.maxIdle = maxIdle;
  106.             return this;
  107.         }
  109.         public MyFlinkJedisClusterConfig.Builder setMinIdle(int minIdle) {
  110.             this.minIdle = minIdle;
  111.             return this;
  112.         }
  114.         public MyFlinkJedisClusterConfig.Builder setPassword(String password) {
  115.             this.password = password;
  116.             return this;
  117.         }
  119.         public MyFlinkJedisClusterConfig build() {
  120.             return new MyFlinkJedisClusterConfig(this.nodes, this.timeout, this.soTimeout,
  121.                     this.maxRedirections, this.maxTotal, this.maxIdle, this.minIdle, this.password);
  122.         }
  123.     }
  124. }
复制代码
燃鹅，中国红却提醒我：
  1. Caused by: java.lang.IllegalArgumentException: Jedis configuration not found
复制代码
原来，Flink任务执行的时候会调用RedisSink中的open()方法：
  1.     public void open(Configuration parameters) throws Exception {
  2.         this.redisCommandsContainer = RedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);
  3.     }
复制代码
而这个方法调用的"RedisCommandsContainerBuilder.build"方法，所使用的参数，依然是旧的FlinkJedisClusterConfig类：
  1.     public static RedisCommandsContainer build(FlinkJedisConfigBase flinkJedisConfigBase)
复制代码
所以，还得改写这两个类：
MyRedisSink：
  1. package org.cube.flink;
  3. /**
  4. * @Author : Lawrence
  5. * @Date : 2022/7/25 23:52
  6. * @Description :
  7. * @Version : 1.0.0
  8. * @Modification_Record :
  9. * Version  Date       Remark
  10. * v1.0.0   2022/7/25  First Create
  11. */
  12. import org.apache.flink.configuration.Configuration;
  13. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  14. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
  15. import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainer;
  16. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
  17. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
  18. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
  19. import org.apache.flink.util.Preconditions;
  20. import org.slf4j.Logger;
  21. import org.slf4j.LoggerFactory;
  23. import java.io.IOException;
  25. public class MyRedisSink<IN> extends RichSinkFunction<IN> {
  26.     private static final long serialVersionUID = 1L;
  27.     private static final Logger LOG = LoggerFactory.getLogger(org.apache.flink.streaming.connectors.redis.RedisSink.class);
  28.     private String additionalKey;
  29.     private RedisMapper<IN> redisSinkMapper;
  30.     private RedisCommand redisCommand;
  31.     private FlinkJedisConfigBase flinkJedisConfigBase;
  32.     private RedisCommandsContainer redisCommandsContainer;
  34.     public MyRedisSink(FlinkJedisConfigBase flinkJedisConfigBase, RedisMapper<IN> redisSinkMapper) {
  35.         Preconditions.checkNotNull(flinkJedisConfigBase, "Redis connection pool config should not be null");
  36.         Preconditions.checkNotNull(redisSinkMapper, "Redis Mapper can not be null");
  37.         Preconditions.checkNotNull(redisSinkMapper.getCommandDescription(), "Redis Mapper data type description can not be null");
  38.         this.flinkJedisConfigBase = flinkJedisConfigBase;
  39.         this.redisSinkMapper = redisSinkMapper;
  40.         RedisCommandDescription redisCommandDescription = redisSinkMapper.getCommandDescription();
  41.         this.redisCommand = redisCommandDescription.getCommand();
  42.         this.additionalKey = redisCommandDescription.getAdditionalKey();
  43.     }
  45.     @Override
  46.     public void invoke(IN input) throws Exception {
  47.         String key = this.redisSinkMapper.getKeyFromData(input);
  48.         String value = this.redisSinkMapper.getValueFromData(input);
  49.         switch(this.redisCommand) {
  50.             case RPUSH:
  51.                 this.redisCommandsContainer.rpush(key, value);
  52.                 break;
  53.             case LPUSH:
  54.                 this.redisCommandsContainer.lpush(key, value);
  55.                 break;
  56.             case SADD:
  57.                 this.redisCommandsContainer.sadd(key, value);
  58.                 break;
  59.             case SET:
  60.                 this.redisCommandsContainer.set(key, value);
  61.                 break;
  62.             case PFADD:
  63.                 this.redisCommandsContainer.pfadd(key, value);
  64.                 break;
  65.             case PUBLISH:
  66.                 this.redisCommandsContainer.publish(key, value);
  67.                 break;
  68.             case ZADD:
  69.                 this.redisCommandsContainer.zadd(this.additionalKey, value, key);
  70.                 break;
  71.             case HSET:
  72.                 this.redisCommandsContainer.hset(this.additionalKey, key, value);
  73.                 break;
  74.             default:
  75.                 throw new IllegalArgumentException("Cannot process such data type: " + this.redisCommand);
  76.         }
  78.     }
  80.     @Override
  81.     public void open(Configuration parameters) throws Exception {
  82.         this.redisCommandsContainer = MyRedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);
  83.     }
  85.     @Override
  86.     public void close() throws IOException {
  87.         if (this.redisCommandsContainer != null) {
  88.             this.redisCommandsContainer.close();
  89.         }
  91.     }
  92. }
复制代码
MyRedisCommandsContainerBuilder：
  1. package org.cube.flink;
  3. /**
  4. * @Author : Lawrence
  5. * @Date : 2022/7/25 21:30
  6. * @Description : 包含了password的RedisCommandsContainerBuilder
  7. * @Version : 1.0.0
  8. * @Modification_Record :
  9. * Version  Date       Remark
  10. * v1.0.0   2022/7/25  First Create
  11. */
  12. import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
  13. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
  14. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
  15. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;
  16. import org.apache.flink.streaming.connectors.redis.common.container.RedisClusterContainer;
  17. import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainer;
  18. import org.apache.flink.streaming.connectors.redis.common.container.RedisContainer;
  19. import org.apache.flink.util.Preconditions;
  20. import redis.clients.jedis.JedisCluster;
  21. import redis.clients.jedis.JedisPool;
  22. import redis.clients.jedis.JedisSentinelPool;
  24. public class MyRedisCommandsContainerBuilder {
  25.     public MyRedisCommandsContainerBuilder() {
  26.     }
  28.     public static RedisCommandsContainer build(FlinkJedisConfigBase flinkJedisConfigBase) {
  29.         if (flinkJedisConfigBase instanceof FlinkJedisPoolConfig) {
  30.             FlinkJedisPoolConfig flinkJedisPoolConfig = (FlinkJedisPoolConfig)flinkJedisConfigBase;
  31.             return build(flinkJedisPoolConfig);
  32.         } else if (flinkJedisConfigBase instanceof MyFlinkJedisClusterConfig) {
  33.             MyFlinkJedisClusterConfig flinkJedisClusterConfig = (MyFlinkJedisClusterConfig)flinkJedisConfigBase;
  34.             return build(flinkJedisClusterConfig);
  35.         } else if (flinkJedisConfigBase instanceof FlinkJedisSentinelConfig) {
  36.             FlinkJedisSentinelConfig flinkJedisSentinelConfig = (FlinkJedisSentinelConfig)flinkJedisConfigBase;
  37.             return build(flinkJedisSentinelConfig);
  38.         } else {
  39.             throw new IllegalArgumentException("Jedis configuration not found");
  40.         }
  41.     }
  43.     public static RedisCommandsContainer build(FlinkJedisPoolConfig jedisPoolConfig) {
  44.         Preconditions.checkNotNull(jedisPoolConfig, "Redis pool config should not be Null");
  45.         GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
  46.         genericObjectPoolConfig.setMaxIdle(jedisPoolConfig.getMaxIdle());
  47.         genericObjectPoolConfig.setMaxTotal(jedisPoolConfig.getMaxTotal());
  48.         genericObjectPoolConfig.setMinIdle(jedisPoolConfig.getMinIdle());
  49.         JedisPool jedisPool = new JedisPool(genericObjectPoolConfig, jedisPoolConfig.getHost(), jedisPoolConfig.getPort()
  50.                 , jedisPoolConfig.getConnectionTimeout(), jedisPoolConfig.getPassword(), jedisPoolConfig.getDatabase());
  51.         return new RedisContainer(jedisPool);
  52.     }
  54.     public static RedisCommandsContainer build(MyFlinkJedisClusterConfig jedisClusterConfig) {
  55.         Preconditions.checkNotNull(jedisClusterConfig, "Redis cluster config should not be Null");
  56.         GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
  57.         genericObjectPoolConfig.setMaxIdle(jedisClusterConfig.getMaxIdle());
  58.         genericObjectPoolConfig.setMaxTotal(jedisClusterConfig.getMaxTotal());
  59.         genericObjectPoolConfig.setMinIdle(jedisClusterConfig.getMinIdle());
  60.         JedisCluster jedisCluster;
  61.         if (null == jedisClusterConfig.getPassword()) {
  62.             jedisCluster = new JedisCluster(jedisClusterConfig.getNodes(), jedisClusterConfig.getConnectionTimeout(),
  63.                     jedisClusterConfig.getMaxRedirections(), genericObjectPoolConfig);
  64.         } else
  65.             {
  66.             jedisCluster = new JedisCluster(jedisClusterConfig.getNodes(), jedisClusterConfig.getConnectionTimeout()
  67.                     , jedisClusterConfig.getSoTimeout(), jedisClusterConfig.getMaxRedirections()
  68.                     , jedisClusterConfig.getPassword(), genericObjectPoolConfig);
  69.         }
  70.         return new RedisClusterContainer(jedisCluster);
  71.     }
  73.     public static RedisCommandsContainer build(FlinkJedisSentinelConfig jedisSentinelConfig) {
  74.         Preconditions.checkNotNull(jedisSentinelConfig, "Redis sentinel config should not be Null");
  75.         GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
  76.         genericObjectPoolConfig.setMaxIdle(jedisSentinelConfig.getMaxIdle());
  77.         genericObjectPoolConfig.setMaxTotal(jedisSentinelConfig.getMaxTotal());
  78.         genericObjectPoolConfig.setMinIdle(jedisSentinelConfig.getMinIdle());
  79.         JedisSentinelPool jedisSentinelPool = new JedisSentinelPool(jedisSentinelConfig.getMasterName()
  80.                 , jedisSentinelConfig.getSentinels(), genericObjectPoolConfig, jedisSentinelConfig.getConnectionTimeout()
  81.                 , jedisSentinelConfig.getSoTimeout(), jedisSentinelConfig.getPassword(), jedisSentinelConfig.getDatabase());
  82.         return new RedisContainer(jedisSentinelPool);
  83.     }
  84. }
复制代码
燃鹅，在重写"MyRedisCommandsContainerBuilder"类时，你会惊奇地发现，jedisCluster 也不支持密码。
你可千万别惯性思维去重新jedisCluster ，
因为这回可真的是版本问题了。
所以这依然难不倒我，
只需要把redis.clients包升级到2.9以上版本即可：
  1.                     redis.clients            jedis            2.9.1                <dependency>
  2.             <groupId>org.apache.flink</groupId>
  3.             <artifactId>flink-connector-redis_2.11</artifactId>
  4.             <version>1.1.5</version>
  5.         </dependency>                                        redis.clients                    jedis                                    
复制代码
好了，到这里咱们终于大功告成了。
 
代码写完了，但是咱们却留下一个疑惑，
为什么这么简单的需求却没有jar包更新呢？
我只是想把Flink数据写到带密码的Redis集群里，这过分吗？
这并不过分，那这又是为啥呢？
我想可能是这样的：
首先，先想一个问题，在流计算中我们往Redis写的是什么数据？
通常是一些状态信息，中间结果。而Flink本身支持状态、缓存和广播机制，导致对Redis的需求下降了。
其次，大数据应用实际运行的环境通常是提交到内网的机器上进行的，大家知道大数据集群之间的主机是需要设置免验证登录的，单单给Redis设密码显得有一点点多余。
其三，Redis的密码机制据说是很弱鸡的，出于安全考虑，更多地是通过防火墙来限制端口，所以很多Redis集群处于管理方便并没有设置密码的。
 其四，出于人类懒惰的本性，发现RedisSink不支持密码后，最省事的方式，或许是放弃使用密码。
 
好了该写的写了，该想的也想了，差不多可以愉快地结束这一天了。
那么晚安了，咱们下期再肝。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

正序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

愛在花開的季節

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表