Как группировать по временному интервалу в Spark SQL

Ответ 1

Искрa >= 2.0

Вы можете использовать window (чтобы не ошибиться с функциями окна). В зависимости от варианта он присваивает временную метку еще одному потенциально перекрывающемуся ведеру:

df.groupBy($"KEY", window($"time", "5 minutes")).sum("metric")

// +---+---------------------------------------------+-----------+
// |KEY|window                                       |sum(metric)|
// +---+---------------------------------------------+-----------+
// |001|[2016-05-01 10:50:00.0,2016-05-01 10:55:00.0]|45         |
// |001|[2016-05-01 10:55:00.0,2016-05-01 11:00:00.0]|12         |
// |003|[2016-05-01 10:55:00.0,2016-05-01 11:00:00.0]|13         |
// |001|[2016-05-01 11:00:00.0,2016-05-01 11:05:00.0]|11         |
// |002|[2016-05-01 10:50:00.0,2016-05-01 10:55:00.0]|100        |
// +---+---------------------------------------------+-----------+

Spark & ​​lt; 2.0

Давайте начнем с данных примера:

import spark.implicits._  // import sqlContext.implicits._ in Spark < 2.0

val df = Seq(
  ("001", "event1", 10, "2016-05-01 10:50:51"),
  ("002", "event2", 100, "2016-05-01 10:50:53"),
  ("001", "event3", 20, "2016-05-01 10:50:55"),
  ("001", "event1", 15, "2016-05-01 10:51:50"),
  ("003", "event1", 13, "2016-05-01 10:55:30"),
  ("001", "event2", 12, "2016-05-01 10:57:00"),
  ("001", "event3", 11, "2016-05-01 11:00:01")
).toDF("KEY", "Event_Type", "metric", "Time")

Я предполагаю, что событие идентифицируется KEY. Если это не так, вы можете настроить предложения GROUP BY/PARTITION BY в соответствии с вашими требованиями.

Если вас интересует агрегация со статическим окном, независимым от данных, преобразуйте временные метки в числовые и круглые

import org.apache.spark.sql.functions.{round, sum}

// cast string to timestamp
val ts = $"Time".cast("timestamp").cast("long")

// Round to 300 seconds interval
val interval = (round(ts / 300L) * 300.0).cast("timestamp").alias("interval")

df.groupBy($"KEY", interval).sum("metric")

// +---+---------------------+-----------+
// |KEY|interval             |sum(metric)|
// +---+---------------------+-----------+
// |001|2016-05-01 11:00:00.0|11         |
// |001|2016-05-01 10:55:00.0|12         |
// |001|2016-05-01 10:50:00.0|45         |
// |003|2016-05-01 10:55:00.0|13         |
// |002|2016-05-01 10:50:00.0|100        |
// +---+---------------------+-----------+

Если вы заинтересованы в окне относительно текущих функций использования строки:

import org.apache.spark.sql.expressions.Window

// Partition by KEY
// Order by timestamp 
// Consider window of -150 seconds to + 150 seconds relative to the current row
val w = Window.partitionBy($"KEY").orderBy("ts").rangeBetween(-150, 150)
df.withColumn("ts", ts).withColumn("window_sum", sum($"metric").over(w))

// +---+----------+------+-------------------+----------+----------+
// |KEY|Event_Type|metric|Time               |ts        |window_sum|
// +---+----------+------+-------------------+----------+----------+
// |003|event1    |13    |2016-05-01 10:55:30|1462092930|13        |
// |001|event1    |10    |2016-05-01 10:50:51|1462092651|45        |
// |001|event3    |20    |2016-05-01 10:50:55|1462092655|45        |
// |001|event1    |15    |2016-05-01 10:51:50|1462092710|45        |
// |001|event2    |12    |2016-05-01 10:57:00|1462093020|12        |
// |001|event3    |11    |2016-05-01 11:00:01|1462093201|11        |
// |002|event2    |100   |2016-05-01 10:50:53|1462092653|100       |
// +---+----------+------+-------------------+----------+----------+

По соображениям производительности этот подход полезен, только если данные могут разбиваться на несколько отдельных групп. В Spark & ​​lt; 2.0.0 вам также понадобится HiveContext, чтобы он работал.

Ответ 2

Для статической границы вы можете выполнить следующие действия:

1) Преобразование (map, mapPartitions и т.д.) Значение времени для формирования YYYY-MM-DD-hh-mm, где mm свертывается с шагом 5 минут. например 01, 02, 03, 05 становится 05; 16,17,18,19,20 становится 20

2) Выполните groupBy или reduceBy с event_type и временем и выполните свою агрегацию (Sum) по меткам

3) Выполните преобразование фильтра для метрик фильтрa > 5

Вы можете написать выше в spark rdd или dataframe (sql) почти так же.

Для другого типа границы, где 00-05, 01-06, 02-07, вы должны попытаться взглянуть на концепцию скользящего окна. Если ваш пример использования для приема данных подходит для потокового шаблона, тогда API Spark Streaming будет идеальным, иначе вы можете найти собственное решение, подобное этому: Apache Spark - Работа с раздвижными окнами на временных RDD