运行Spark流作业时出现序列化问题

Question

无法解决以下由filtered.foreachpartition（iter=>{）触发的序列化问题。我认为foreachpartition可以解决序列化问题，但事实并非如此。那么，如何使用redispool？

val redis_host = "localhost"
val redist_port = 6379
messages.foreachRDD(rdd => {
  rdd.foreachPartition(iter => {
    val redisPool = new Pool(new JedisPool(new JedisPoolConfig(), redis_host, redis_port, 2000))
    iter.foreach({ msg =>
      println(msg.mkString(","))
    })
  })
})

我假设变量redis_host和redis_port是不可序列化的，但是我如何正确地序列化它们，以便代码能够在集群上工作，而不仅仅是在本地工作呢？

上面显示的代码抛出错误：

Answer 1

解决方案是在匿名函数中懒洋洋地初始化池。在java中,可以执行以下操作：

messages.foreachRDD(new RedisFunction(redis_host, redis_port))
messages.count()

class RedisFunction implements F {
    private Pool pool = null;
    private final String redis_host;
    private final String redis_port;

    RedisFunction(String redis_host, String redis_port) {
        this.redis_host = redis_host;
        this.redis_port = redis_port;
        initPool();
    }
    private void initPool() {
        this.pool = new Pool(new JedisPool(new JedisPoolConfig(), redis_host, redis_port, 2000))
    }
    public Void call(JavaRDD<> rdd) {
        if(this.pool == null) {
            initPool();
        }
        rdd = rdd.map(....);  /*your rdd transformations go here*/
        rdd.count();   //spark action
    }
}

上面的java示例应该可以帮助您解决序列化问题。