Spark - «частичная» оконная функция

Question

В Spark, пытающейся выполнить «частичную» оконную функцию.

Предоставление схемы eventId, impressionTime, кампании, дохода , я хотел бы знать для каждого вечера сумму дохода длякампания за последние 4 дня.

В конце я добавляю только последние 3 дня.Поэтому мне пришлось загрузить данные за 7 дней (поэтому у самого раннего события есть окно 4 дня), и проблема в том, что я буду рассчитывать также и события из моего окончательного диапазона.

Небольшой пример: у меня есть данныев течение 7 дней (день1, день2, день3, день4, день5, день6, день7 7 позже) и в выходных данных только дни: требуется 7,6,5.

Мне нужно окно для каждых 4дней.Поэтому я буду загружать все дни и делать для них оконную функцию.В конце я буду длиться всего 3 дня.

Конечно, это очень неэффективно, так как я рассчитываю окно также для дней 4,3 ...

Есть ли в любом случае функция окна для некоторых данных?

Спасибо

Answer 1

Оконные функции не предоставляют метод, с помощью которого вы можете вычислять агрегаты только для подмножества дат ваших входных данных (дни 5, 6 и 7). И как вы указали, вы не можете предварительно фильтровать, потому что данные от дней 1 до 4 необходимы для построения точных совокупностей окон.

Если вы хотите использовать оконные функции Spark, вам нужно будет отфильтровать дни с 1 по 4 после выполнения агрегации окон.

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

val df1 = spark.read.table("table1").filter(col("day").isin(Seq(5,6,7):_*))
val df2 = spark.read.table("table1").filter(col("day").isin(Seq(1,2,3,4,5,6,7):_*))

df1.
  join(df2, (df1("key1") === df2("key2")).and(df2("millisecond_ts").between(df1("millisecond_ts") - 1000*60*60*24*4, df1("millisecond_ts"))), "left")