由于 Jedis Pool，Spark 程序卡住了答案

【问题标题】：Spark program stucks because of Jedis Pool由于 Jedis Pool，Spark 程序卡住了
【发布时间】：2021-09-26 15:45:37
【问题描述】：

根据我的previous question 使用 Jedis 和线程。我将代码更改为使用JedisPool 而不是Jedis。但是仍然随着线程的增加程序卡住了。我试图增加.setMaxIdle(8000) 和.setMaxTotal(8000) 并临时修复它，但后来在其他运行中它在一些迭代后再次卡住。我猜是由于池中缺少连接（我关闭了它们），但似乎线程没有释放连接。

这是我的连接的更新版本：

import redis.clients.jedis.{JedisPool, JedisPoolConfig}
 
object redisOp{
  @transient lazy val log: Logger = org.apache.log4j.LogManager.getLogger("myLogger")
  def apply(set: RDD[Int]): Unit = {
    val spark = SparkConstructor()
    val sc = spark.sparkContext
    // initialize Parents and Ranks key-values
    val parents = set.map(i => ("p"+i, i.toString))
    val ranks = set.map(i => ("r"+i, 1.toString))
    sc.toRedisKV(parents) // using spark-redis packege here only, ignore it.
    sc.toRedisKV(ranks)
    log.warn("***Initialized Redis***")
 
  }
 
  val jedisConfig = new JedisPoolConfig()          // Check from here (object's values and variables)
  jedisConfig.setMaxIdle(8000)                    //TODO: a better configuration?
  jedisConfig.setMaxTotal(8000)
  lazy val pool = new JedisPool(jedisConfig, "localhost")

  def find(u: Long): Option[Long] = { // returns leader of the set containing u
    val r = pool.getResource
    val res = Option(r.get(s"p$u")).flatMap(p => if (p.toLong == u) {
      Some(u)
    } else find(p.toLong))
    r.close()                                    // closing back to pool
    res
  }
// other methods are similar to find()...
}

【问题讨论】：

标签： scala apache-spark redis connection-pooling jedis

【解决方案1】：

问题在于您的递归实现。您正在调用下一个递归堆栈而不释放资源。因此，在某些时候，最新的堆栈处于资源稀缺状态，因为旧堆栈占用了它们的资源。

所以，在调用下一个递归栈之前释放资源。

例如

  def find(u: Long): Option[Long] = { // returns leader of the set containing u
    val r = pool.getResource
    val rget = r.get(s"p$u")
    r.close()                                    // closing back to pool
    val res = Option(rget).flatMap(p => if (p.toLong == u) {
      Some(u)
    } else find(p.toLong))
    res
  }

【讨论】：