2016-04-06 27 views
7

我写过一个方法,必须考虑一个随机数来模拟伯努利分布。我正在使用random.nextDouble生成0到1之间的数字,然后根据给定我的概率参数的值作出我的决定。Spark - Random Number Generation

我的问题是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 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)像下面,但我仍然得到在每一个同一个号码循环

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

我做了一些研究,看起来这与Sparks的确定性本质有关。

回答

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) 
} 
+0

我修改这个稍微解决我的问题。我将Random val传入我的方法,并从那里生成随机数。这解决了我的问题,但出于序列化原因,我不得不使用java.util.Random'。 –

4

只需使用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| 
+---+------------------+ 
+0

这并没有完全解决我的问题,但其优雅的解决方案,我可能会被使用在未来,所以+1 –

2

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))) 
}