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

问题描述

我正在尝试在单例 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()

  }

}

解决方法

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

apache-spark connection-pool scala scala singleton