Spark - Génération De Nombres Aléatoires

J'ai écrit une méthode qui doit tenir compte d'un nombre aléatoire pour simuler une distribution de Bernoulli. Je suis à l'aide de random.nextDouble pour générer un nombre entre 0 et 1 puis rendre ma décision sur la base de cette valeur compte tenu de mon probabilité paramètre.

Mon problème est que l'Étincelle est de générer les nombres aléatoires à l'intérieur de chaque itération de ma boucle for fonction de mappage. Je suis l'aide de la DataFrame API. Mon code suit ce format:

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

Ici, c'est la classe:

class myClass extends Serializable {
  val q = qProb

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

J'ai besoin d'un nouveau nombre aléatoire à chaque fois myMethod est appelé. J'ai aussi essayé de générer le nombre à l'intérieur de ma méthode avec java.util.Random (scala.util.Random v10 ne s'étend pas Serializable) comme ci-dessous, mais je suis toujours avoir le même nombre à l'intérieur de chaque boucle

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

J'ai fait quelques recherches, et il semble que cela a à faire avec des Étincelles déterministe de la nature.

OriginalL'auteur Brian Vanover | 2016-04-06

  1. 3

    La raison pour laquelle la même séquence est répétée, c'est que le générateur aléatoire est créé et initialisé avec une graine avant que les données sont partitionnées. Chaque partition, puis commence à partir de la même graine. Peut-être pas le moyen le plus efficace de le faire, mais la suivante devrait fonctionner: 

    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)
}
    J'ai modifié un peu pour résoudre mon problème. J'ai passé l'Aléatoire val dans ma méthode et de nombres aléatoires générés à partir de l'intérieur. Cela a résolu mon problème, mais j'ai dû utiliser java.util.Random pour serializeability raisons.

    OriginalL'auteur Pascal Soucy

  2. 8

    Suffit d'utiliser la fonction 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|
+---+------------------+
    Ce n'est pas tout résoudre mon problème, mais c'est une solution élégante, je vais probablement utiliser dans le futur, donc +1

    OriginalL'auteur David Griffin

  3. 3

    Selon ce post, la meilleure solution est de ne pas mettre la new scala.util.Random à l'intérieur de la carte, ni complètement à l'extérieur (ie. dans le code du driver), mais à un intermédiaire 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)))
}

    OriginalL'auteur leo9r