Как определить функцию WINDOWING в SQL-запросе Spark, чтобы избежать повторяющегося кода - PullRequest
Новая
       46

Как определить функцию WINDOWING в SQL-запросе Spark, чтобы избежать повторяющегося кода

0 голосов
/ 08 мая 2018

У меня есть запрос, который имеет много опережений и задержек, из-за чего повторяется разделение по коду.

Если я использую код Scala, я могу определить спецификацию окна и использовать ее повторно, поэтому есть способ, которым я могу повторно использовать код раздела в Spark SQL.

Цель состоит в том, чтобы избежать повторения "over (разбиение по sessionId, упорядочение устройства по идентификатору entry_datetime)" 

SELECT * ,
lag( channel,1,null ) over ( partition by sessionId, deviceId order by entry_datetime ) as prev_chnl,
lead( channel,1,null ) over ( partition by sessionId, deviceId order by entry_datetime ) as next_chnl,
lag( channel-source,1,null ) over ( partition by sessionId, deviceId order by entry_datetime ) as prev_chnl_source,
lead( channel-source,1,null ) over ( partition by sessionId, deviceId order by entry_datetime ) as next_chnl_source,
FROM RAW_VIEW

RAW_VIEW 

+------------+-----------+---------------------+---------+-----------------+
|sessionId   |deviceId   |entry_datetime       |channel  |channel-source   |
+------------+-----------+---------------------+---------+-----------------+
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 15:00:00.0|001      |Internet         |
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 16:00:00.0|002      |Cable            |
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 17:00:00.0|003      |Satellite        |
+------------+-----------+---------------------+---------+-----------------+

ЗАКЛЮЧИТЕЛЬНЫЙ ВИД 

+------------+-----------+---------------------+---------+-----------------+---------+---------+-----------------+-----------------+
|sessionId   |deviceId   |entry_datetime       |channel  |channel-source   |prev_chnl|next_chnl|prev_chnl_source |next_chnl_source |
+------------+-----------+---------------------+---------+-----------------+---------+---------+-----------------+-----------------+
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 15:00:00.0|001      |Internet         |null     |002      |null             |Cable            |
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 15:01:00.0|002      |Cable            |001      |003      |Internet         |Satellite        |
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 15:02:00.0|003      |Satellite        |002      |null     |Cable            |null             |
+------------+-----------+---------------------+---------+-----------------+---------+---------+-----------------+-----------------+

Ответы [ 2 ]

0 голосов
/ 08 мая 2018

Вы должны быть в состоянии определить именованное окно и ссылаться на него в запросе: 

SELECT * ,
  lag(channel, 1) OVER w AS prev_chnl,
  lead(channel, 1) OVER w AS next_chnl,
  lag(channel-source, 1) OVER w AS prev_chnl_source,
  lead(channel-source, 1) OVER w AS next_chnl_source,
FROM raw_view
WINDOW w AS (PARTITION BY sessionId, deviceId ORDER BY entry_datetime)

но похоже, что эта функция в данный момент не работает.

0 голосов
/ 08 мая 2018

Если вы хотите сделать это в spark-sql, одним из способов является добавление row_number() к вашей таблице поверх ваших заказанных разделов. Затем создайте версию этой таблицы с задержкой и опережением, вычитая / добавляя 1 к номеру строки. Наконец, выполните LEFT JOIN текущей таблицы с предыдущей и следующей версиями и выберите соответствующие столбцы.

Например, попробуйте следующее: 

   SELECT curr.*,
          prev.channel AS prev_chnl,
          next.channel AS next_chnl,
          prev.channel_source AS prev_chnl_source,
          next.channel_source AS next_chnl_source
     FROM (SELECT *,
                 ROW_NUMBER() OVER (partition by sessionId, 
                                                  deviceId 
                                        order by entry_datetime) AS row_num 
           FROM RAW_VIEW
     ) curr
LEFT JOIN (SELECT *,
                  ROW_NUMBER() OVER (partition by sessionId,
                                                  deviceId
                                         order by entry_datetime) + 1 AS row_num
           FROM RAW_VIEW
     ) prev ON (curr.row_num = prev.row_num)
LEFT JOIN (SELECT *,
                  ROW_NUMBER() OVER (partition by sessionId,
                                                  deviceId
                                         order by entry_datetime) - 1 AS row_num
           FROM RAW_VIEW
     ) next ON (next.row_num = curr.row_num)
 ORDER BY entry_datetime

Что приводит к: 

+------------+-----------+---------------------+-------+--------------+-------+---------+---------+----------------+----------------+
|sessionId   |deviceId   |entry_datetime       |channel|channel_source|row_num|prev_chnl|next_chnl|prev_chnl_source|next_chnl_source|
+------------+-----------+---------------------+-------+--------------+-------+---------+---------+----------------+----------------+
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 15:00:00.0|001    |Internet      |1      |null     |002      |null            |Cable           |
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 16:00:00.0|002    |Cable         |2      |001      |003      |Internet        |Satellite       |
|SESSION-ID-1|DEVICE-ID-1|2018-04-09 17:00:00.0|003    |Satellite     |3      |002      |null     |Cable           |null            |
+------------+-----------+---------------------+-------+--------------+-------+---------+---------+----------------+----------------+
...