在 Scala 中向单例对象添加配置

Question

我正在尝试在 Singleton Scala 对象中设置到 Redis 的连接池，以便在映射 DF 的分区时可以 read/write 到 Redis。当我运行我的主要方法时，我希望能够配置主机以及其他连接池变量。但是，当前的配置没有给我配置的 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()

  }

}

Answer 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 执行任何其他操作之前设置它，您应该是安全的。

在 Scala 中向单例对象添加配置

Add Configurations to a Singleton Object in Scala

singleton

scala

apache-spark

connection-pool