2017-01-11 173 views
2

出於評估目的生成數據集(或數據幀),我需要創建一個功能的dummy- Dataset(或者一個DataFrame),用隨機數初始化。在列和行方面的尺寸應被參數星火和斯卡拉:給定大小

我想出了一個解決方案,但是這是荒謬的慢(5.3s爲10行100列):

def createDummyDataset(rows : Int, columns: Int, spark: SparkSession) = { 
    import spark.implicits._ 

    var ds = Seq.fill(rows)(Random.nextDouble).toDF() 
    if (columns > 1) { 
     for (i <- 2 to columns) { 
     ds = ds.withColumn(i.toString, rand) 
     } 
    } 
    ds // return ds 
} 

是,由於Spark的架構,還是我在做一些完全錯誤的事情,還有更好的方法?

我想一個更好的辦法是定義某種矩陣,轉換至一個Dataset在打擊。但我無法弄清楚。

系統:星火2.1.0,斯卡拉2.11.8,Ubuntu的16.04,i5-6300U,32GB的RAM

回答

1

通過添加列到現有的數據幀否則它會引起很多星火相關的開銷。

更能營造然後一個二維數組式的集合並行所有一氣呵成:

import org.apache.spark.sql.Row 
import spark.implicits._ 

val data = (0 to rows).map(_ => Seq.fill(columns)(Random.nextDouble)) 
val rdd = sc.parallelize(data) 
val df = rdd.map(s => Row.fromSeq(s)).toDF() 
+0

謝謝。你試過了嗎?它減少了10行和100列的執行時間,以3.3s(-61%)。但是,這仍然可以延長1000個電池的使用時間嗎? – Boern

+1

是似乎一點就慢側。什麼是你定時 - 整個星火作業,或者將數據幀的只是初始化?初始化Seq的次序應該是次秒。我不希望Spark將這個大小的數據集分發超過3秒。多久沒有考慮到執行(0至行).MAP(_ => Seq.fill(列)(Random.nextDouble))。toDF()。rdd.collect()? – ImDarrenG

+0

我正在初始化時間...此外,您的代碼導致一個'DataFrame'只有一個列持有一個數組數組:( – Boern

1

基於ImDarrenG答案,但輸出爲n行m列的數據幀。

import org.apache.spark.sql.{Row, SparkSession} 
import org.apache.spark.sql.types.{DoubleType, StructField, StructType} 

def start(rows: Int, cols: Int, col: String, spark: SparkSession): Unit = { 

     val data = (1 to rows).map(_ => Seq.fill(cols)(Random.nextDouble)) 

     val colNames = (1 to cols).mkString(",") 
     val sch = StructType(colNames.split(",").map(fieldName => StructField(fieldName, DoubleType, true))) 

     val rdd = spark.sparkContext.parallelize(data.map(x => Row(x:_*))) 
     val df = spark.sqlContext.createDataFrame(rdd, sch) 

     df.printSchema() 

     spark.stop() 
    } 

運行在火花2.1.0,Scala的2.11.8,Fedora的科學,i5-5200U 4個芯,16GB RAM

對於10行×100列的平均經過時間爲0.9秒

+0

[Boern](http://stackoverflow.com/users/1701600/boern),這是否回答你的問題? – geo