在Spark DataFrame中佈置TimeSeries數據的最佳方式 - 斯卡拉

Question

我從美聯儲經濟數據集API導入數據。每個請求會返回每日，每週，每月或每年的時間序列。我的最終目標是做變量選擇並建立一個基於貝葉斯的模型，使用選定的時間序列作爲特定時間序列的預測變量。將這些數據組織成數據框架的最佳方式是什麼？

根據這documentation我想我的數據應該列在「Instants」format。然而，我嘗試加入超過200,000列的結果後，所有這些嘗試都變得極其緩慢。下面的文檔參考中的另一種格式是「TimeSeriesRDD」，但導入的時間序列通常沒有日期重疊，範圍從1930年到現在。那麼，將這些數據組織成數據框架的最佳方式是什麼？

如何將FRED中的數據加載到推薦的格式中的示例將非常感謝！

這是我的第一種方法是慢得

for (seriesId <- allSeries) { 
    val series = loadSeriesFromAPI(seriesId, spark) 
    allSeries = allSeries.join(series, allSeries.col("date") === series.col(seriesId + "_date"), "outer") 
    allSeries = allSeries.drop(seriesId + "_date") 
} 

我的第二個，我將不得不在一個時間

for(row <- series) { 
    val insertStr = "%s, %g". 
    format(
     row.asInstanceOf[Map[String, Date]]("date").asInstanceOf[String], 
     parseDoubleOrZero(row.asInstanceOf[Map[String, Double]]("value").asInstanceOf[String]) 
    ) 
} 

Answer 1

有一個數據幀中的數據加載1列和1列200.000列不是一個好主意。有一兩件事，我建議是分裂的問題有點不混合太多的技術：

1. 數據攝取：有多大的系列實際上是誰？儘可能避免加入（加入意味着洗牌，洗牌意味着網絡，這將使一切變得緩慢）。我會用Scala收集數據並將其保存在內存中，如果不合適，我仍然會在Scala中收集批量的系列，並將每個批量轉換爲Spark DataFrame。
2. 數據幀創建：如果你成功地獲得內存數據，那麼你可以試試下面的代碼片段將創建你一個數據幀：

case class Point(timestamp: Long, value: Long) 
case class Series(id: String, points: List[Point]) 

val s1 = Series("s1", List(Point(1, 100), Point(2, 200), Point(3, 100))) 
val s2 = Series("s2", List(Point(1, 1000), Point(3, 100))) 

val seriesDF = sc.parallelize(Array(s1, s2)).toDF 
seriesDF.show() 

seriesDF.select($"id", explode($"points").as("point")) 
    .select($"id", $"point.timestamp", $"point.value") 
    .show() 

輸出：

+---+--------------------+ 
| id|    points| 
+---+--------------------+ 
| s1|[[1,100], [2,200]...| 
| s2| [[1,1000], [3,100]]| 
+---+--------------------+ 
+---+---------+-----+ 
| id|timestamp|value| 
+---+---------+-----+ 
| s1|  1| 100| 
| s1|  2| 200| 
| s1|  3| 100| 
| s2|  1| 1000| 
| s2|  3| 100| 
+---+---------+-----+ 

對於更花哨的方式來處理時間序列我會推薦以下項目：https://github.com/twosigma/flint