Хранение большого количества данных временных рядов в Scala Dataframe с общим ключом

Question

Я пытаюсь хранить более 200 000 переменных временных рядов в кадре данных Spark, и я хочу, чтобы даты для каждого временного ряда совпадали с набором первичных ключей всех дней между 1-1-1930 и 1-1-2017. Конечной целью является создание байесовской модели, которая использует множество временных рядов в качестве предикторов для определенного временного ряда. Я прочитал временный ряд XML в искровом Dataframe, который выглядит следующим образом:

+---------------------+----------------------+ 
|timeSeries1_date  |timeSeries1_value  | 
+---------------------+----------------------+ 
|   1997-01-01|     10.49| 
|   1997-01-02|     15.64| 
|   1997-01-03|     53.06| 
|     ....|     ....| 
+---------------------+----------------------+ 

Прямо сейчас я пытаюсь сделать полную внешнее объединение, чтобы добавить каждый дополнительный временный ряд с общим набором временных рядов:

allSeries = allSeries.join(series, allSeries.col("date") === newSeries.col("timeSeries2_date"), "outer") 
allSeries = allSeries.drop("timeSeries2_date") 

Результат выглядит следующим образом:

+---------------------+----------------------+----------------------+ 
|date     |timeSeries1_value  |timeSeries2_value  | 
+---------------------+----------------------+----------------------+ 
|   1997-01-01|     10.49|     11.93| 
|   1997-01-02|     15.64|     null| 
|   1997-01-03|     53.06|     43.04| 
|     ....|     ....|     ....| 
+---------------------+----------------------+----------------------+ 

Это присоединиться становится крайне медленно до точки она является непригодным для использования. Мой опыт в основном связан с реляционными базами данных, поэтому у меня возникают проблемы с пониманием того, какой правильный подход будет для этой ситуации. Нужно ли мне даже объединять эти данные в один и тот же Dataframe? Есть ли более эффективный способ сделать это?

Answer 1

Вам следует избегать использования кросс-соединений на больших наборах данных - как вы упомянули, это слишком медленно.

Вы могли бы решить проблему путем группировки данных по дате (с временной зернистостью вам нужно - dayofmonth, hour, month или year), а затем применить функцию агрегации с aggAPI или пользовательским UDAF. Например, если ваша задача состояла в том, чтобы вычислить среднюю суточную цену для следующего кадра данных:

scala> data.show() 
+----------+-------+ 
| timestamp| price| 
+----------+-------+ 
|1997-01-01| 10.49| 
|1997-01-01|1150.63| 
|1997-01-02| 15.64| 
|1997-01-03| 53.06| 
+----------+-------+ 

Вы можете достичь его следующим образом:

scala> data.groupBy(dayofmonth(col("timestamp")).alias("day")).agg(mean(col("price")).alias("mean_daily_price")).show() 
+----+-----------------+ 
|day |mean_daily_price | 
+----+-----------------+ 
| 1|580.5600000000001| 
| 3|   53.06| 
| 2|   15.64| 
+----+-----------------+