Лидерство и отставание в групповом режиме

Question

Для каждого entity_id в наборе данных сигналов найдите item_id с самым старым и самым новым month_id.

В некоторых случаях это может быть один и тот же элемент.Если есть 2 разных предмета с одинаковым month_id, возьмите предмет с меньшим item_id.

Наконец, суммируйте количество сигналов для каждого объекта и выведите как total_signals.Правильный вывод должен содержать 1 строку для каждого уникального entity_id.

Ввод:

entity_id: long
item_id: integer
source: integer
month_id: integer
signal_count: integer

Выход:

entity_id: long
oldest_item_id: integer
newest_item_id: integer
total_signals: integer

Как использовать оконные функции опережающие и запаздывающие столбцы month_id?

input:

+ --------- + -------+ ------ + -------- + ------------ +

| entity_id | item_id | source | month_id | signal_count |

+ --------- + ------- + ------ + -------- + ------------+

|359781 |20001 |21000 |201705 |1 |

|359781 |20001 |21000 |201707 |1 |

|359781 |3 |21000 |201708 |2 |

|359781 |3 |21000 |201712 |2 |

|359781 |3 |975 |201512 |4 |

вывод:

entity_id Старейший_item_id

с использованием фрейма данных scala api и spark

Answer 1

Вы бы не использовали lead() и lag() для этого.Вы бы использовали агрегацию.В этом случае вам также потребуется некоторая условная агрегация.

Это может быть лучшим подходом:

select t.entity_id, t.num_signals,
       tmin.item_id as item_id_min_month,
       tmax.item_id as item_id_max_month
from (select t.entity_id, sum(signal_count) as num_signals,
             min(month_id) as min_month_id,
             max(month_id) as max_month_id
      from t
     ) t join
     t tmin
     on tmin.entity_id = t.entity_id and
        tmin.month_id = t.min_month_id join
     t tmax
     on tmax.entity_id = t.entity_id and
        tmax.month_id = t.max_month_id ;

Вы также можете использовать first_value() и last_value(), если они доступны вваша версия SparkSQL.