1

我正在尝试在 Singleton Scala 对象中设置到 Redis 的连接池，以便在映射 DF 的分区时可以读取/写入 Redis。当我运行我的 main 方法时，我希望能够配置主机以及其他连接池变量。但是，这个当前配置并没有给我我配置的 REDIS_HOST，它给了我localhost。

在写这篇文章时，我引用了https://able.bio/patrickcording/sharing-objects-in-spark--58x4gbf每个执行者一个实例部分。

在为每个执行程序维护一个 RedisClient 实例的同时实现配置主机的最佳方法是什么？

object Main {
  def main(args: Array[String]): Unit = {
    val parsedConfig = ConfigFactory.parseFile(new File(args(0)))
    val config = ConfigFactory.load(parsedConfig)
    RedisClient.host = config.getString("REDIS_HOST")
    val Main = new Main()
    Main.runMain()
  }
}

class Main{
    val df = Seq(...).toDF()
    df.mapPartitions(partitions => {
        partitions.foreach(row => {
           val count =  RedisClient.getIdCount(row.getAs("id").asInstanceOf[String])
            //do something
        })
    })

    df.write.save
    RedisClient.close()
}

object RedisClient {
  var host: String = "localhost"

  private val pool = new RedisClientPool(host, 6379)

  def getIdCount(id: String):Option[String] = {
    pool.withClient(client => {
      client.get(orderLineId)
    })
  }

  def close(): Unit = {
    pool.close()

  }

}
4

1 回答 1

1

在 Spark 中，main仅在驱动程序上运行，而不是在执行程序上运行。 RedisClient在您调用调用它的方法之前，不能保证在任何给定的执行程序上都存在，并且它只会使用默认值进行初始化。

因此，确保它拥有正确主机的唯一方法是，在相同的 RDD/DF 操作中，确保host已设置，例如：

df.mapPartitions(partitions => {
  RedisClient.host = config.getString("REDIS_HOST")
  partitions.foreach(row => {
    ...
  }
}

当然，由于main不在驱动程序上运行，您可能还希望将配置广播给执行程序：

// after setting up the SparkContext
val sc: SparkContext = ???
val broadcastConfig = sc.broadcast(config)

然后您将传递broadcastConfig并使用broadcastConfig.value代替config，因此上述内容将变为：

df.mapPartitions(partitions => {
  RedisClient.host = broadcastConfig.value.getString("REDIS_HOST")
  partitions.foreach(row => {
    ...
  }
}

只要您注意始终将相同的值分配给RedisClient.host并在对 进行任何其他操作之前对其进行设置RedisClient，您就应该是安全的。

于 2021-02-12T21:28:45.207 回答