Spark - 随机数生成

Question

我写了一个必须考虑随机数的方法来模拟伯努利分布。我正在使用 random.nextDouble 生成一个介于 0 和 1 之间的数字，然后根据我的概率参数根据该值做出决定。

我的问题是 Spark 在我的 for 循环映射函数的每次迭代中生成相同的随机数。我正在使用 DataFrame API。我的代码遵循以下格式：

val myClass = new MyClass()
val M = 3
val myAppSeed = 91234
val rand = new scala.util.Random(myAppSeed)

for (m <- 1 to M) {
  val newDF = sqlContext.createDataFrame(myDF
    .map{row => RowFactory
      .create(row.getString(0),
        myClass.myMethod(row.getString(2), rand.nextDouble())
    }, myDF.schema)
}

这里是 class:

class myClass extends Serializable {
  val q = qProb

  def myMethod(s: String, rand: Double) = {
    if (rand <= q) // do something
    else // do something else
  }
}

每次调用myMethod我都需要一个新的随机数。我还尝试使用 java.util.Random（scala.util.Random v10 不扩展 Serializable）在我的方法中生成数字，如下所示，但我仍然在每个 for 循环中得到相同的数字

val r = new java.util.Random(s.hashCode.toLong)
val rand = r.nextDouble()

我做了一些研究，这似乎与 Sparks 的确定性有关。

Answer 1

只需使用SQL函数rand:

import org.apache.spark.sql.functions._

//df: org.apache.spark.sql.DataFrame = [key: int]

df.select($"key", rand() as "rand").show
+---+-------------------+
|key|               rand|
+---+-------------------+
|  1| 0.8635073400704648|
|  2| 0.6870153659986652|
|  3|0.18998048357873532|
+---+-------------------+


df.select($"key", rand() as "rand").show
+---+------------------+
|key|              rand|
+---+------------------+
|  1|0.3422484248879837|
|  2|0.2301384925817671|
|  3|0.6959421970071372|
+---+------------------+

Answer 2

重复相同序列的原因是随机生成器是在数据分区之前用种子创建和初始化的。然后每个分区从相同的随机种子开始。也许不是最有效的方法，但以下方法应该有效：

val myClass = new MyClass()
val M = 3

for (m <- 1 to M) {
  val newDF = sqlContext.createDataFrame(myDF
    .map{ 
       val rand = scala.util.Random
       row => RowFactory
      .create(row.getString(0),
        myClass.myMethod(row.getString(2), rand.nextDouble())
    }, myDF.schema)
}

Answer 3

根据this post，最好的解决方案不是将new scala.util.Random放在地图内部，也不是完全放在地图外部（即在驱动程序代码中），而是在中间mapPartitionsWithIndex:

import scala.util.Random
val myAppSeed = 91234
val newRDD = myRDD.mapPartitionsWithIndex { (indx, iter) =>
   val rand = new scala.util.Random(indx+myAppSeed)
   iter.map(x => (x, Array.fill(10)(rand.nextDouble)))
}

Answer 4

使用 Spark 数据集 API，可能用于累加器：

df.withColumn("_n", substring(rand(),3,4).cast("bigint"))

Spark - 随机数生成

Spark - Random Number Generation

random

scala

apache-spark

spark-dataframe