我如何計算殘差的相關性？ Spark-Scala

Question

我需要知道我的殘差是否相關。我沒有找到在Databricks上使用Spark-Scala的方法。 而我的結論是，我應該將我的項目導出到R使用acf function。

有人知道在Databricks上使用Spark-Scala的技巧嗎？

對於那些需要更多信息的人：我目前正在進行銷售預測。我使用了不同的功能的迴歸森林。然後，我需要評估我的預測質量。爲了檢查這一點，我讀了這paper殘差是一個很好的方式來看看你的預測模型是好還是壞。在任何情況下，您仍然可以改進它，但只是對我的預測模型發表我的看法，並將其與其他模型進行比較。

目前，我有一個像下面的數據框。它是測試數據/樣本外數據的一部分。 （我鑄造預測殘差IntegerType，這就是爲什麼在第3行40 - 17 = 22）

Answer 1

我使用Spark 2.1.1。

可以使用spark ml library function

找到列之間的關係讓我們首先導入的類。

import org.apache.spark.sql.functions.corr 
import org.apache.spark.mllib.linalg._ 
import org.apache.spark.mllib.stat.Statistics 

現在準備輸入數據框：

scala> val seqRow = Seq(
    |  ("2017-04-27",13,21), 
    |  ("2017-04-26",7,16), 
    |  ("2017-04-25",40,17), 
    |  ("2017-04-24",17,17), 
    |  ("2017-04-21",10,20), 
    |  ("2017-04-20",9,19), 
    |  ("2017-04-19",30,30), 
    |  ("2017-04-18",18,25), 
    |  ("2017-04-14",32,28), 
    |  ("2017-04-13",39,18), 
    |  ("2017-04-12",2,4), 
    |  ("2017-04-11",8,24), 
    |  ("2017-04-10",18,27), 
    |  ("2017-04-07",6,17), 
    |  ("2017-04-06",13,29), 
    |  ("2017-04-05",10,17), 
    |  ("2017-04-04",6,8), 
    |  ("2017-04-03",20,32) 
    |) 
seqRow: Seq[(String, Int, Int)] = List((2017-04-27,13,21), (2017-04-26,7,16), (2017-04-25,40,17), (2017-04-24,17,17), (2017-04-21,10,20), (2017-04-20,9,19), (2017-04-19,30,30), (2017-04-18,18,25), (2017-04-14,32,28), (2017-04-13,39,18), (2017-04-12,2,4), (2017-04-11,8,24), (2017-04-10,18,27), (2017-04-07,6,17), (2017-04-06,13,29), (2017-04-05,10,17), (2017-04-04,6,8), (2017-04-03,20,32)) 

scala> val rdd = sc.parallelize(seqRow) 
rdd: org.apache.spark.rdd.RDD[(String, Int, Int)] = ParallelCollectionRDD[51] at parallelize at <console>:34 

scala> val input_df = spark.createDataFrame(rdd).toDF("date","amount","prediction").withColumn("residuals",'amount - 'prediction) 
input_df: org.apache.spark.sql.DataFrame = [date: string, amount: int ... 2 more fields] 

scala> input_df.show(false) 
+----------+------+----------+---------+ 
|date  |amount|prediction|residuals| 
+----------+------+----------+---------+ 
|2017-04-27|13 |21  |-8  | 
|2017-04-26|7  |16  |-9  | 
|2017-04-25|40 |17  |23  | 
|2017-04-24|17 |17  |0  | 
|2017-04-21|10 |20  |-10  | 
|2017-04-20|9  |19  |-10  | 
|2017-04-19|30 |30  |0  | 
|2017-04-18|18 |25  |-7  | 
|2017-04-14|32 |28  |4  | 
|2017-04-13|39 |18  |21  | 
|2017-04-12|2  |4   |-2  | 
|2017-04-11|8  |24  |-16  | 
|2017-04-10|18 |27  |-9  | 
|2017-04-07|6  |17  |-11  | 
|2017-04-06|13 |29  |-16  | 
|2017-04-05|10 |17  |-7  | 
|2017-04-04|6  |8   |-2  | 
|2017-04-03|20 |32  |-12  | 
+----------+------+----------+---------+ 

爲2017-04-14行和2017-04-13不匹配的residuals值，因爲我減去amount - prediction爲residuals

現在在進行着計算相關在所有列之間。 如果列數更多，並且您需要每列與其他列之間的相關性，則此方法用於計算相關性。

首先我們放棄之間的相關度是不被計算

scala> val drop_date_df = input_df.drop('date) 
drop_date_df: org.apache.spark.sql.DataFrame = [amount: int, prediction: int ... 1 more field] 

scala> drop_date_df.show 
+------+----------+---------+ 
|amount|prediction|residuals| 
+------+----------+---------+ 
| 13|  21|  -8| 
|  7|  16|  -9| 
| 40|  17|  23| 
| 17|  17|  0| 
| 10|  20|  -10| 
|  9|  19|  -10| 
| 30|  30|  0| 
| 18|  25|  -7| 
| 32|  28|  4| 
| 39|  18|  21| 
|  2|   4|  -2| 
|  8|  24|  -16| 
| 18|  27|  -9| 
|  6|  17|  -11| 
| 13|  29|  -16| 
| 10|  17|  -7| 
|  6|   8|  -2| 
| 20|  32|  -12| 
+------+----------+---------+ 

由於有超過2列的相關性，我們需要找到相關矩陣的列。 對於計算相關矩陣我們需要RDD [矢量]正如您可以在火花示例中看到的相關性。

scala> val dense_rdd = drop_date_df.rdd.map{row => 
    |   val first = row.getAs[Integer]("amount") 
    |   val second = row.getAs[Integer]("prediction") 
    |   val third = row.getAs[Integer]("residuals") 
    |   Vectors.dense(first.toDouble,second.toDouble,third.toDouble)} 
dense_rdd: org.apache.spark.rdd.RDD[org.apache.spark.mllib.linalg.Vector] = MapPartitionsRDD[62] at map at <console>:40 

scala> val correlMatrix: Matrix = Statistics.corr(dense_rdd, "pearson") 
correlMatrix: org.apache.spark.mllib.linalg.Matrix = 
1.0     0.40467032516705076 0.782939330961529 
0.40467032516705076 1.0     -0.2520531290688281 
0.782939330961529 -0.2520531290688281 1.0 

列的順序保持不變，但是您排除了列名稱。 您可以找到關於相關矩陣結構的好資源。

既然你想找到殘差與其他兩列的相關性。 我們可以探索其他選項

蜂巢corr UDAF

scala> drop_date_df.createOrReplaceTempView("temp_table") 

scala> val corr_query_df = spark.sql("select corr(amount,residuals) as amount_residuals_corr,corr(prediction,residuals) as prediction_residual_corr from temp_table") 
corr_query_df: org.apache.spark.sql.DataFrame = [amount_residuals_corr: double, prediction_residual_corr: double] 

scala> corr_query_df.show 
+---------------------+------------------------+ 
|amount_residuals_corr|prediction_residual_corr| 
+---------------------+------------------------+ 
| 0.7829393309615287|  -0.252053129068828| 
+---------------------+------------------------+ 

星火科爾功能link

scala> val corr_df = drop_date_df.select(
    |     corr('amount,'residuals).as("amount_residuals_corr"), 
    |     corr('prediction,'residuals).as("prediction_residual_corr")) 
corr_df: org.apache.spark.sql.DataFrame = [amount_residuals_corr: double, prediction_residual_corr: double] 

scala> corr_df.show 
+---------------------+------------------------+ 
|amount_residuals_corr|prediction_residual_corr| 
+---------------------+------------------------+ 
| 0.7829393309615287|  -0.252053129068828| 
+---------------------+------------------------+