如何生成一个包含随机内容和N行的DataFrame？

Question

如何在 Scala 中创建一个包含 100 行和 3 列且随机整数值在 (1, 100) 范围内的 Spark DataFrame？

我知道如何手动创建 DataFrame，但我无法自动化它：

val df = sc.parallelize(Seq((1,20, 40), (60, 10, 80), (30, 15, 30))).toDF("col1", "col2", "col3") 

Answer 1

在本地生成数据然后并行化它是完全可以的，特别是当您不需要生成大量数据时。

但是，如果您需要生成巨大的数据集，您始终可以实现一个 RDD 来并行执行此操作，如以下示例所示。

import scala.reflect.ClassTag
import org.apache.spark.{Partition, TaskContext}
import org.apache.spark.rdd.RDD

// Each random partition will hold `numValues` items
final class RandomPartition[A: ClassTag](val index: Int, numValues: Int, random: => A) extends Partition {
  def values: Iterator[A] = Iterator.fill(numValues)(random)
}

// The RDD will parallelize the workload across `numSlices`
final class RandomRDD[A: ClassTag](@transient private val sc: SparkContext, numSlices: Int, numValues: Int, random: => A) extends RDD[A](sc, deps = Seq.empty) {

  // Based on the item and executor count, determine how many values are
  // computed in each executor. Distribute the rest evenly (if any).
  private val valuesPerSlice = numValues / numSlices
  private val slicesWithExtraItem = numValues % numSlices

  // Just ask the partition for the data
  override def compute(split: Partition, context: TaskContext): Iterator[A] =
    split.asInstanceOf[RandomPartition[A]].values

  // Generate the partitions so that the load is as evenly spread as possible
  // e.g. 10 partition and 22 items -> 2 slices with 3 items and 8 slices with 2
  override protected def getPartitions: Array[Partition] =
    ((0 until slicesWithExtraItem).view.map(new RandomPartition[A](_, valuesPerSlice + 1, random)) ++
      (slicesWithExtraItem until numSlices).view.map(new RandomPartition[A](_, valuesPerSlice, random))).toArray

}

一旦有了这个，您就可以使用它传递您自己的随机数据生成器来获得

RDD[Int]

val rdd = new RandomRDD(spark.sparkContext, 10, 22, scala.util.Random.nextInt(100) + 1)
rdd.foreach(println)
/*
 * outputs:
 * 30
 * 86
 * 75
 * 20
 * ...
 */

或

RDD[(Int, Int, Int)]

def rand = scala.util.Random.nextInt(100) + 1
val rdd = new RandomRDD(spark.sparkContext, 10, 22, (rand, rand, rand))
rdd.foreach(println)
/*
 * outputs:
 * (33,22,15)
 * (65,24,64)
 * (41,81,44)
 * (58,7,18)
 * ...
 */

当然你也可以很容易地将它包裹在

DataFrame

中：

spark.createDataFrame(rdd).show()
/*
 * outputs:
 * +---+---+---+
 * | _1| _2| _3|
 * +---+---+---+
 * |100| 48| 92|
 * | 34| 40| 30|
 * | 98| 63| 61|
 * | 95| 17| 63|
 * | 68| 31| 34|
 * .............
 */

请注意，在这种情况下，每次执行

RDD

/

DataFrame

时生成的数据都是不同的。通过更改

RandomPartition

的实现来实际存储值而不是动态生成它们，您可以拥有一组稳定的随机项，同时仍然保留此方法的灵活性和可扩展性。

无状态方法的一个很好的特性是您甚至可以在本地生成巨大的数据集。以下内容在我的笔记本电脑上运行了几秒钟：

new RandomRDD(spark.sparkContext, 10, Int.MaxValue, 42).count
// returns: 2147483647

Answer 2

给你，

Seq.fill

是你的朋友：

def randomInt1to100 = scala.util.Random.nextInt(100)+1

val df = sc.parallelize(
  Seq.fill(100){(randomInt1to100,randomInt1to100,randomInt1to100)}
).toDF("col1", "col2", "col3")

Answer 3

您可以简单地使用

scala.util.Random

生成范围内的 随机数 并循环 100 行，最后使用

createDataFrame

api

import scala.util.Random
val data = 1 to 100 map(x =>  (1+Random.nextInt(100), 1+Random.nextInt(100), 1+Random.nextInt(100)))

sqlContext.createDataFrame(data).toDF("col1", "col2", "col3").show(false)

Answer 4

您可以使用下面的通用代码

//no of rows required
val rows = 15
//no of columns required
val cols = 10

val spark = SparkSession.builder
  .master("local[*]")
  .appName("testApp")
  .config("spark.sql.warehouse.dir", "file:///c:/tmp/spark-warehouse")
  .getOrCreate()

import spark.implicits._

val columns = 1 to cols map (i => "col" + i)

// create the DataFrame schema with these columns (in that order)
val schema = StructType(columns.map(StructField(_, IntegerType)))

val lstrows = Seq.fill(rows * cols)(Random.nextInt(100) + 1).grouped(cols).toList.map { x => Row(x: _*) }

val rdd = spark.sparkContext.makeRDD(lstrows)
val df = spark.createDataFrame(rdd, schema)

Answer 5

如果您需要创建大量随机数据，Spark 提供了一个名为 RandomRDDs 的对象，它可以生成填充有遵循均匀分布、正态分布或各种其他分布的随机数的数据集。

https://spark.apache.org/docs/latest/mllib-statistics.html#random-data- Generation

从他们的例子来看：

import org.apache.spark.mllib.random.RandomRDDs._

// Generate a random double RDD that contains 1 million i.i.d. values drawn from the
// standard normal distribution `N(0, 1)`, evenly distributed in 10 partitions.
val u = normalRDD(sc, 1000000L, 10)
// Apply a transform to get a random double RDD following `N(1, 4)`.
val v = u.map(x => 1.0 + 2.0 * x)

Answer 6

从 Spark 1.6 开始，已经有了隐式的 toDF，可以轻松地从任何

Seq

或

Range

创建 DataFrame

val spark: SparkSession = ???
import spark.sqlContext.implicits._

val df = Range(1, 100).toDF("n")

创建后，您可以放弃范围列并选择您喜欢的任何其他内容。

val df = Range(1, 100).toDF().select(
  rand() as "x",
  when($"x" < 0.5, lit(true)) otherwise lit(false) as "y"
).persist()

使用持久只是为了停止随机重新评估。