Эффективный способ генерировать большие рандомизированные данные в Spark

Question

Я пытаюсь создать большой случайный набор данных искры. По сути, я хочу начать с 2018-12-01 09:00:00, и для каждой новой строки отметка времени изменится на scala.util.Random.nextInt(3) секунды. (Столбец timestamp является единственным значимым столбцом)

Я хочу, чтобы это продолжало работать, даже когда я пытаюсь сгенерировать триллионы строк в большом кластере, поэтому я пытаюсь сгенерировать его партиями по 100 элементов за раз, поскольку триллионы строк не могут вписаться в Seq .

Есть несколько проблем с этим кодом, например var, и я не уверен в том, что я использую union. Мне интересно, есть ли у кого-нибудь лучшее представление о том, как это сделать.

import Math.{max, min}
import java.sql.Timestamp
import java.sql.Timestamp.valueOf

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Row, SaveMode}
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

object DataGenerator extends SparkEnv {

  import spark.implicits._

  val batchSize = 100
  val rnd = scala.util.Random

  // randomly generates a DataFrame with n Rows
  def generateTimestampData(n: Int): DataFrame = {
    val timestampDataFields = Seq(StructField("timestamp", TimestampType, false))
    val initDF = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], StructType(timestampDataFields))
    def loop(data: DataFrame, lastTime: Long, _n: Int): DataFrame = {
      if (_n == 0) {
        val w = Window.orderBy("timestamp")
        data.withColumn("eventID", concat(typedLit("event"), row_number().over(w)))
      } else {
        var thisTime = lastTime
        def rts(ts: Long): Stream[Long] = ts #:: { thisTime = ts + rnd.nextInt(3) * 1000; rts(thisTime) }
        val thisBatch = rts(lastTime)
          .map(new Timestamp(_))
          .take(min(batchSize, _n))
          .toDF("timestamp")
        loop(data union thisBatch, thisTime, max(_n - batchSize, 0))
      }
    }
    loop(initDF, valueOf("2018-12-01 09:00:00").getTime(), n)
  }

  def main(args: Array[String]): Unit = {
    val w = Window.orderBy("timestamp")
    val df = generateTimestampData(10015)
      .withColumn("part", floor(row_number().over(w) / 100))
    df.repartition(27)
      .write
      .partitionBy("part")
      .option("compression", "snappy")
      .mode(SaveMode.Overwrite)
      .parquet("data/generated/ts_data")
  }

}

Приведенный выше код приводит к созданию DataFrame с 10 015 строками, который выглядит примерно так:

+-------------------+----------+----+
|          timestamp|   eventID|part|
+-------------------+----------+----+
|2018-12-01 11:43:09|event10009| 100|
|2018-12-01 11:43:02|event10003| 100|
|2018-12-01 11:43:11|event10012| 100|
|2018-12-01 11:43:10|event10011| 100|
|2018-12-01 11:43:08|event10007| 100|
|2018-12-01 11:43:02|event10001| 100|
|2018-12-01 11:43:08|event10008| 100|
|2018-12-01 11:43:12|event10013| 100|
|2018-12-01 11:43:09|event10010| 100|
|2018-12-01 11:43:14|event10014| 100|
|2018-12-01 10:11:54| event4357|  43|
|2018-12-01 10:47:33| event6524|  65|
|2018-12-01 10:23:08| event5064|  50|
|2018-12-01 10:23:02| event5060|  50|
|2018-12-01 10:23:39| event5099|  50|
|2018-12-01 10:22:25| event5019|  50|
|2018-12-01 09:16:36| event1042|  10|
|2018-12-01 09:16:03| event1008|  10|
|2018-12-01 09:16:13| event1017|  10|
|2018-12-01 09:17:28| event1092|  10|
+-------------------+----------+----+

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