火花柱狀表演

我是一個相對初學者Spark的東西。我有一個廣泛的數據幀（1000列），我想將列添加到基於對應的列是否有遺漏值火花柱狀表演

所以

 
+----+   
| A | 
+----+ 
| 1 | 
+----+ 
|null|  
+----+ 
| 3 | 
+----+

成爲

 
+----+-------+   
| A | A_MIS | 
+----+-------+ 
| 1 | 0 | 
+----+-------+ 
|null| 1 | 
+----+-------+ 
| 3 | 1 | 
+----+-------+

這是一部分定製ml變壓器，但算法應該清晰。

override def transform(dataset: org.apache.spark.sql.Dataset[_]): org.apache.spark.sql.DataFrame = { 
    var ds = dataset 
    dataset.columns.foreach(c => { 
    if (dataset.filter(col(c).isNull).count() > 0) { 
     ds = ds.withColumn(c + "_MIS", when(col(c).isNull, 1).otherwise(0)) 
    } 
    }) 


    ds.toDF() 
}

循環遍歷列，如果> 0個空值創建一個新列。

傳入的數據集被緩存（使用.cache方法），相關配置設置是默認值。現在在單檯筆記本電腦上運行，即使使用最少量的行，也可以在1000列上運行40分鐘。我認爲這個問題是由於碰到一個數據庫造成的，所以我試着用parquet文件來取代相同的結果。看看作業用戶界面，它似乎在做文件掃描以便進行計數。

有沒有一種方法可以改進此算法以獲得更好的性能，或以某種方式調整緩存？增加spark.sql.inMemoryColumnarStorage.batchSize剛剛給我一個OOM錯誤。

來源

2017-09-13 BinderNet

刪除條件：

if (dataset.filter(col(c).isNull).count() > 0)

，只留下內部表達。正如它所寫的，Spark需要#columns數據掃描。

如果您希望修剪列一次計算統計信息，如Count number of non-NaN entries in each column of Spark dataframe with Pyspark中所述，並使用單個drop調用。

來源

2017-09-13 08:17:29 user8371915

謝謝。嘗試了這種方法，並得到了2分鐘，這是我可以接受的。我會在下面發佈我的代碼。 – BinderNet

下面是修復問題的代碼。

override def transform(dataset: Dataset[_]): DataFrame = { 
    var ds = dataset 
    val rowCount = dataset.count() 
    val exprs = dataset.columns.map(count(_)) 
    val colCounts = dataset.agg(exprs.head, exprs.tail: _*).toDF(dataset.columns: _*).first() 
    dataset.columns.foreach(c => { 
    if (colCounts.getAs[Long](c) > 0 && colCounts.getAs[Long](c) < rowCount ) { 
     ds = ds.withColumn(c + "_MIS", when(col(c).isNull, 1).otherwise(0)) 
    } 
    }) 
    ds.toDF() 
}

來源

2017-09-13 10:26:55 BinderNet

火花柱狀表演

回答

相關問題