Flink写入Redis集群重写flink-connector-redis包，解决Cluster无法输入密码问题

起因：使用flink的时候难免和redis打交道，相信大家都使用过flink-connector-redis来处理，但是当我想要使用RedisSink写入集群时，发现居然不支持使用密码，于是有了这篇笔记。

事情的经过是这样的，我准备用Flink往Redis写入数据，我照常引入flink-connector-redis包

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-connector-redis_2.11</artifactId>

            <version>1.1.5</version>

        </dependency>

然后洋洋洒洒写下如下代码：

package org.cube.flink

import org.apache.flink.streaming.api.scala.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, RedisCommandDescription, RedisMapper}

import java.net.InetSocketAddress

import java.util.HashSet

/**

 * @Author : Lawrence

 * @Date : 2022/7/24 23:11

 * @Description : Flink结果写入Redis集群

 * @Version : 1.0.0

 * @Modification_Record:

 * Version  Date       Remark

 * v1.0.0   2022/7/24  First Create

 */

object RedisClusterSink {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    // source

    import org.apache.flink.api.scala._

    val source = env.fromElements("1 hadoop","2 spark","3 flink", "4 hive", "5 redis", "6 hbase")

    // process

    val tupleValue = source.map(_.split(" ")).map(x => (x(0), x(1)))

    // redis config

    val builder = new FlinkJedisPoolConfig.Builder

    builder.setHost("cube01").setPort(7001).setPassword("123456")

    val redisConf: FlinkJedisPoolConfig = builder.build()

    // sink

    val redisSink = new RedisSink[(String, String)](redisConf, new MyRedisMapper())

    tupleValue.addSink(redisSink)

    env.execute("RedisClusterSink")

  }

}

class MyRedisMapper extends RedisMapper[(String, String)] {

  override def getCommandDescription: RedisCommandDescription = {

    new RedisCommandDescription(RedisCommand.SET)

  }

  override def getKeyFromData(t: (String, String)): String = t._1

  override def getValueFromData(t: (String, String)): String = t._2

}

然后兴高采烈地点击了运行，控制台却给了我一抹中国红，

其中最后一条是这样说的：

Caused by: redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 9842 192.168.20.132:7003

哦哦，是因为我的Redis是集群模式，

这并难不倒我，

我只需要把FlinkJedisPoolConfig改成FlinkJedisClusterConfig就万事大吉了。

    // redis config

    val builder = new FlinkJedisClusterConfig.Builder

    val inetSocketAddress = new InetSocketAddress("cube01", 7001)

    val nodeSet = new HashSet[InetSocketAddress]()

    nodeSet.add(inetSocketAddress)

    builder.setNodes(nodeSet).setPassword("123456")

    val redisConf: FlinkJedisClusterConfig = builder.build()

可是，这个类并没有setPassword方法，事实上它连"password"这个属性都没有。

这并难不倒我。

先不设密码总行了吧？

燃鹅并不行，控制台又给了我一抹中国红，

他是这样说的：

Caused by: redis.clients.jedis.exceptions.JedisDataException: NOAUTH Authentication required.

呵呵，这可难不倒我，

我的本能反应是，应该到Maven仓库中找到新版的flink-connector-redis包。

燃鹅，当我搜索之后发现，这已经是最新版了。

这也难不倒我。

FlinkJedisPoolConfig不是可以设置密码吗？

FlinkJedisClusterConfig不是可以访问集群吗？

如果我把他们两个的代码整合一下呢？是不是就好了。

于是我本能地把"FlinkJedisClusterConfig"改写成了"MyFlinkJedisClusterConfig"类，增加了password属性和对应的get，set方法。

package org.cube.flink;

/**

 * @Author : Lawrence

 * @Date : 2022/7/25 21:14

 * @Description : 包含了password的FlinkJedisClusterConfig

 * @Version : 1.0.0

 * @Modification_Record:

 * Version  Date       Remark

 * v1.0.0   2022/7/25  First Create

 */

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;

import org.apache.flink.util.Preconditions;

import redis.clients.jedis.HostAndPort;

import redis.clients.jedis.Protocol;

import java.net.InetSocketAddress;

import java.util.HashSet;

import java.util.Iterator;

import java.util.Set;

public class MyFlinkJedisClusterConfig extends FlinkJedisConfigBase {

    private static final long serialVersionUID = 1L;

    private final Set<InetSocketAddress> nodes;

    private final int maxRedirections;

    private int soTimeout;

    private String password;

    private MyFlinkJedisClusterConfig(Set<InetSocketAddress> nodes, int connectionTimeout, int soTimeout,

                                      int maxRedirections, int maxTotal, int maxIdle, int minIdle, String password) {

        super(connectionTimeout, maxTotal, maxIdle, minIdle);

        Preconditions.checkNotNull(nodes, "Node information should be presented");

        Preconditions.checkArgument(!nodes.isEmpty(), "Redis cluster hosts should not be empty");

        this.nodes = new HashSet(nodes);

        this.soTimeout = soTimeout;

        this.maxRedirections = maxRedirections;

        this.password = password;

    }

    public Set<HostAndPort> getNodes() {

        Set<HostAndPort> ret = new HashSet();

        Iterator var2 = this.nodes.iterator();

        while(var2.hasNext()) {

            InetSocketAddress node = (InetSocketAddress)var2.next();

            ret.add(new HostAndPort(node.getHostName(), node.getPort()));

        }

        return ret;

    }

    public int getMaxRedirections() {

        return this.maxRedirections;

    }

    public int getSoTimeout() { return this.soTimeout; }

    protected String getPassword() { return this.password; }

    public String toString() {

        return "JedisClusterConfig{nodes=" + this.nodes + ", timeout=" + this.connectionTimeout

                + ", maxRedirections=" + this.maxRedirections + ", maxTotal=" + this.maxTotal

                + ", maxIdle=" + this.maxIdle + ", minIdle=" + this.minIdle + '}';

    }

    public static class Builder {

        private Set<InetSocketAddress> nodes;

        private int timeout = Protocol.DEFAULT_TIMEOUT;

        private int maxRedirections = 5;

        //新增属性

        private int soTimeout = Protocol.DEFAULT_TIMEOUT;

        private int maxTotal = GenericObjectPoolConfig.DEFAULT_MAX_TOTAL;

        private int maxIdle = GenericObjectPoolConfig.DEFAULT_MAX_IDLE;

        private int minIdle = GenericObjectPoolConfig.DEFAULT_MIN_IDLE;

        //增加的属性

        private String password;

        public Builder() {

        }

        public MyFlinkJedisClusterConfig.Builder setNodes(Set<InetSocketAddress> nodes) {

            this.nodes = nodes;

            return this;

        }

        public MyFlinkJedisClusterConfig.Builder setTimeout(int timeout) {

            this.timeout = timeout;

            return this;

        }

        public MyFlinkJedisClusterConfig.Builder setSoTimeout(int soTimeout) {

            this.soTimeout = soTimeout;

            return this;

        }

        public MyFlinkJedisClusterConfig.Builder setMaxRedirections(int maxRedirections) {

            this.maxRedirections = maxRedirections;

            return this;

        }

        public MyFlinkJedisClusterConfig.Builder setMaxTotal(int maxTotal) {

            this.maxTotal = maxTotal;

            return this;

        }

        public MyFlinkJedisClusterConfig.Builder setMaxIdle(int maxIdle) {

            this.maxIdle = maxIdle;

            return this;

        }

        public MyFlinkJedisClusterConfig.Builder setMinIdle(int minIdle) {

            this.minIdle = minIdle;

            return this;

        }

        public MyFlinkJedisClusterConfig.Builder setPassword(String password) {

            this.password = password;

            return this;

        }

        public MyFlinkJedisClusterConfig build() {

            return new MyFlinkJedisClusterConfig(this.nodes, this.timeout, this.soTimeout,

                    this.maxRedirections, this.maxTotal, this.maxIdle, this.minIdle, this.password);

        }

    }

}

燃鹅，中国红却提醒我：

Caused by: java.lang.IllegalArgumentException: Jedis configuration not found

原来，Flink任务执行的时候会调用RedisSink中的open()方法：

    public void open(Configuration parameters) throws Exception {

        this.redisCommandsContainer = RedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);

    }

而这个方法调用的"RedisCommandsContainerBuilder.build"方法，所使用的参数，依然是旧的FlinkJedisClusterConfig类：

    public static RedisCommandsContainer build(FlinkJedisConfigBase flinkJedisConfigBase)

所以，还得改写这两个类：

MyRedisSink：

package org.cube.flink;

/**

 * @Author : Lawrence

 * @Date : 2022/7/25 23:52

 * @Description :

 * @Version : 1.0.0

 * @Modification_Record :

 * Version  Date       Remark

 * v1.0.0   2022/7/25  First Create

 */

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;

import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainer;

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;

import org.apache.flink.util.Preconditions;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.io.IOException;

public class MyRedisSink<IN> extends RichSinkFunction<IN> {

    private static final long serialVersionUID = 1L;

    private static final Logger LOG = LoggerFactory.getLogger(org.apache.flink.streaming.connectors.redis.RedisSink.class);

    private String additionalKey;

    private RedisMapper<IN> redisSinkMapper;

    private RedisCommand redisCommand;

    private FlinkJedisConfigBase flinkJedisConfigBase;

    private RedisCommandsContainer redisCommandsContainer;

    public MyRedisSink(FlinkJedisConfigBase flinkJedisConfigBase, RedisMapper<IN> redisSinkMapper) {

        Preconditions.checkNotNull(flinkJedisConfigBase, "Redis connection pool config should not be null");

        Preconditions.checkNotNull(redisSinkMapper, "Redis Mapper can not be null");

        Preconditions.checkNotNull(redisSinkMapper.getCommandDescription(), "Redis Mapper data type description can not be null");

        this.flinkJedisConfigBase = flinkJedisConfigBase;

        this.redisSinkMapper = redisSinkMapper;

        RedisCommandDescription redisCommandDescription = redisSinkMapper.getCommandDescription();

        this.redisCommand = redisCommandDescription.getCommand();

        this.additionalKey = redisCommandDescription.getAdditionalKey();

    }

    @Override

    public void invoke(IN input) throws Exception {

        String key = this.redisSinkMapper.getKeyFromData(input);

        String value = this.redisSinkMapper.getValueFromData(input);

        switch(this.redisCommand) {

            case RPUSH:

                this.redisCommandsContainer.rpush(key, value);

                break;

            case LPUSH:

                this.redisCommandsContainer.lpush(key, value);

                break;

            case SADD:

                this.redisCommandsContainer.sadd(key, value);

                break;

            case SET:

                this.redisCommandsContainer.set(key, value);

                break;

            case PFADD:

                this.redisCommandsContainer.pfadd(key, value);

                break;

            case PUBLISH:

                this.redisCommandsContainer.publish(key, value);

                break;

            case ZADD:

                this.redisCommandsContainer.zadd(this.additionalKey, value, key);

                break;

            case HSET:

                this.redisCommandsContainer.hset(this.additionalKey, key, value);

                break;

            default:

                throw new IllegalArgumentException("Cannot process such data type: " + this.redisCommand);

        }

    }

    @Override

    public void open(Configuration parameters) throws Exception {

        this.redisCommandsContainer = MyRedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);

    }

    @Override

    public void close() throws IOException {

        if (this.redisCommandsContainer != null) {

            this.redisCommandsContainer.close();

        }

    }

}

MyRedisCommandsContainerBuilder：

package org.cube.flink;

/**

 * @Author : Lawrence

 * @Date : 2022/7/25 21:30

 * @Description : 包含了password的RedisCommandsContainerBuilder

 * @Version : 1.0.0

 * @Modification_Record :

 * Version  Date       Remark

 * v1.0.0   2022/7/25  First Create

 */

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;

import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;

import org.apache.flink.streaming.connectors.redis.common.container.RedisClusterContainer;

import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainer;

import org.apache.flink.streaming.connectors.redis.common.container.RedisContainer;

import org.apache.flink.util.Preconditions;

import redis.clients.jedis.JedisCluster;

import redis.clients.jedis.JedisPool;

import redis.clients.jedis.JedisSentinelPool;

public class MyRedisCommandsContainerBuilder {

    public MyRedisCommandsContainerBuilder() {

    }

    public static RedisCommandsContainer build(FlinkJedisConfigBase flinkJedisConfigBase) {

        if (flinkJedisConfigBase instanceof FlinkJedisPoolConfig) {

            FlinkJedisPoolConfig flinkJedisPoolConfig = (FlinkJedisPoolConfig)flinkJedisConfigBase;

            return build(flinkJedisPoolConfig);

        } else if (flinkJedisConfigBase instanceof MyFlinkJedisClusterConfig) {

            MyFlinkJedisClusterConfig flinkJedisClusterConfig = (MyFlinkJedisClusterConfig)flinkJedisConfigBase;

            return build(flinkJedisClusterConfig);

        } else if (flinkJedisConfigBase instanceof FlinkJedisSentinelConfig) {

            FlinkJedisSentinelConfig flinkJedisSentinelConfig = (FlinkJedisSentinelConfig)flinkJedisConfigBase;

            return build(flinkJedisSentinelConfig);

        } else {

            throw new IllegalArgumentException("Jedis configuration not found");

        }

    }

    public static RedisCommandsContainer build(FlinkJedisPoolConfig jedisPoolConfig) {

        Preconditions.checkNotNull(jedisPoolConfig, "Redis pool config should not be Null");

        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();

        genericObjectPoolConfig.setMaxIdle(jedisPoolConfig.getMaxIdle());

        genericObjectPoolConfig.setMaxTotal(jedisPoolConfig.getMaxTotal());

        genericObjectPoolConfig.setMinIdle(jedisPoolConfig.getMinIdle());

        JedisPool jedisPool = new JedisPool(genericObjectPoolConfig, jedisPoolConfig.getHost(), jedisPoolConfig.getPort()

                , jedisPoolConfig.getConnectionTimeout(), jedisPoolConfig.getPassword(), jedisPoolConfig.getDatabase());

        return new RedisContainer(jedisPool);

    }

    public static RedisCommandsContainer build(MyFlinkJedisClusterConfig jedisClusterConfig) {

        Preconditions.checkNotNull(jedisClusterConfig, "Redis cluster config should not be Null");

        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();

        genericObjectPoolConfig.setMaxIdle(jedisClusterConfig.getMaxIdle());

        genericObjectPoolConfig.setMaxTotal(jedisClusterConfig.getMaxTotal());

        genericObjectPoolConfig.setMinIdle(jedisClusterConfig.getMinIdle());

        JedisCluster jedisCluster;

        if (null == jedisClusterConfig.getPassword()) {

            jedisCluster = new JedisCluster(jedisClusterConfig.getNodes(), jedisClusterConfig.getConnectionTimeout(),

                    jedisClusterConfig.getMaxRedirections(), genericObjectPoolConfig);

        } else

            {

            jedisCluster = new JedisCluster(jedisClusterConfig.getNodes(), jedisClusterConfig.getConnectionTimeout()

                    , jedisClusterConfig.getSoTimeout(), jedisClusterConfig.getMaxRedirections()

                    , jedisClusterConfig.getPassword(), genericObjectPoolConfig);

        }

        return new RedisClusterContainer(jedisCluster);

    }

    public static RedisCommandsContainer build(FlinkJedisSentinelConfig jedisSentinelConfig) {

        Preconditions.checkNotNull(jedisSentinelConfig, "Redis sentinel config should not be Null");

        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();

        genericObjectPoolConfig.setMaxIdle(jedisSentinelConfig.getMaxIdle());

        genericObjectPoolConfig.setMaxTotal(jedisSentinelConfig.getMaxTotal());

        genericObjectPoolConfig.setMinIdle(jedisSentinelConfig.getMinIdle());

        JedisSentinelPool jedisSentinelPool = new JedisSentinelPool(jedisSentinelConfig.getMasterName()

                , jedisSentinelConfig.getSentinels(), genericObjectPoolConfig, jedisSentinelConfig.getConnectionTimeout()

                , jedisSentinelConfig.getSoTimeout(), jedisSentinelConfig.getPassword(), jedisSentinelConfig.getDatabase());

        return new RedisContainer(jedisSentinelPool);

    }

}

燃鹅，在重写"MyRedisCommandsContainerBuilder"类时，你会惊奇地发现，jedisCluster 也不支持密码。

你可千万别惯性思维去重新jedisCluster ，

因为这回可真的是版本问题了。

所以这依然难不倒我，

只需要把redis.clients包升级到2.9以上版本即可：

        <dependency>

            <groupId>redis.clients</groupId>

            <artifactId>jedis</artifactId>

            <version>2.9.1</version>

        </dependency>

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-connector-redis_2.11</artifactId>

            <version>1.1.5</version>

            <exclusions>

                <exclusion>

                    <groupId>redis.clients</groupId>

                    <artifactId>jedis</artifactId>

                </exclusion>

            </exclusions>

        </dependency>

好了，到这里咱们终于大功告成了。

代码写完了，但是咱们却留下一个疑惑，

为什么这么简单的需求却没有jar包更新呢？

我只是想把Flink数据写到带密码的Redis集群里，这过分吗？

这并不过分，那这又是为啥呢？

我想可能是这样的：

首先，先想一个问题，在流计算中我们往Redis写的是什么数据？

通常是一些状态信息，中间结果。而Flink本身支持状态、缓存和广播机制，导致对Redis的需求下降了。

其次，大数据应用实际运行的环境通常是提交到内网的机器上进行的，大家知道大数据集群之间的主机是需要设置免验证登录的，单单给Redis设密码显得有一点点多余。

其三，Redis的密码机制据说是很弱鸡的，出于安全考虑，更多地是通过防火墙来限制端口，所以很多Redis集群处于管理方便并没有设置密码的。

其四，出于人类懒惰的本性，发现RedisSink不支持密码后，最省事的方式，或许是放弃使用密码。

好了该写的写了，该想的也想了，差不多可以愉快地结束这一天了。

那么晚安了，咱们下期再肝。

PS：代码需要可以直接搬运，不过这里只是应急处理，更稳妥的做法是直接改写原来的类重新打包。而不是同时保留两个类似的类，这样容易造成混乱。

巴特西

Flink写入Redis集群重写flink-connector-redis包，解决Cluster无法输入密码问题

最新文章

热门文章

巴特西

Flink写入Redis集群 重写flink-connector-redis包，解决Cluster无法输入密码问题

最新文章

热门文章

Flink写入Redis集群重写flink-connector-redis包，解决Cluster无法输入密码问题