Временные ряды со скалой и искрой. Роллинг окно

Question

Я пытаюсь выполнить следующее упражнение, используя Scala и spark.

Дан файл, содержащий два столбца: время в секундах и значение

Пример:

|---------------------|------------------|
|     seconds         |     value        |
|---------------------|------------------|
|          225        |         1,5      |
|          245        |         0,5      |
|          300        |         2,4      |
|          319        |         1,2      |
|          320        |         4,6      |
|---------------------|------------------|

и с учетом значения V, которое будет использоваться для скользящего окна, должен быть создан этот вывод:

Пример с V=20

|--------------|---------|--------------------|----------------------|
|     seconds  |  value  |  num_row_in_window |sum_values_in_windows |
|--------------|---------|--------------------|----------------------|
|       225    |    1,5  |          1         |          1,5         |
|       245    |    0,5  |          2         |           2          |
|       300    |    2,4  |          1         |          2,4         |
|       319    |    1,2  |          2         |          3,6         |
|       320    |    4,6  |          3         |          8,2         |
|--------------|---------|--------------------|----------------------|

num_row_in_window - количество строк, содержащихся в текущем окне, и sum_values_in_windows - сумма значений, содержащихся в текущем окне.

Я пробовал использовать функцию скольжения или использовать sql api, но мне немного непонятно, какое решение лучше всего решить, учитывая, что я новичок в spark / scala.

Answer 1

Это идеальное приложение для оконных функций. Используя rangeBetween, вы можете установить скользящее окно на 20 с. Обратите внимание, что в приведенном ниже примере разделение не указано (не partitionBy). Без разделения этот код не будет масштабироваться:

import ss.implicits._

val df = Seq(
  (225, 1.5),
  (245, 0.5),
  (300, 2.4),
  (319, 1.2),
  (320, 4.6)
).toDF("seconds", "value")

val window = Window.orderBy($"seconds").rangeBetween(-20L, 0L) // add partitioning here

df
  .withColumn("num_row_in_window", sum(lit(1)).over(window))
  .withColumn("sum_values_in_window", sum($"value").over(window))
  .show()

+-------+-----+-----------------+--------------------+
|seconds|value|num_row_in_window|sum_values_in_window|
+-------+-----+-----------------+--------------------+
|    225|  1.5|                1|                 1.5|
|    245|  0.5|                2|                 2.0|
|    300|  2.4|                1|                 2.4|
|    319|  1.2|                2|                 3.6| 
|    320|  4.6|                3|                 8.2|
+-------+-----+-----------------+--------------------+