Элегантно объединяющиеся строки в Spark, основанные на нескольких условиях - PullRequest
Новая
       4

Элегантно объединяющиеся строки в Spark, основанные на нескольких условиях

0 голосов
/ 30 марта 2020

Heyo StackOverflow,

В настоящее время я пытаюсь найти элегантный способ выполнить конкретное c преобразование.

Итак, у меня есть блок данных действий, который выглядит следующим образом: 

+---------+----------+----------+---------+
|timestamp|   user_id|    action|    value|
+---------+----------+----------+---------+
|      100|         1|     click|     null|
|      101|         2|     click|     null|
|      103|         1|      drag|      AAA|
|      101|         1|     click|     null|
|      108|         1|     click|     null|
|      100|         2|     click|     null|
|      106|         1|      drag|      BBB|
+---------+----------+----------+---------+

Контекст: пользователи могут выполнять действия: щелчки и перетаскивания. Клики не имеют значения, перетаскивают. Перетаскивание подразумевает, что был щелчок, но не наоборот. Также предположим, что событие перетаскивания может быть записано после или до события щелчка. Таким образом, у меня есть для каждого перетаскивания соответствующее действие щелчка. То, что я хотел бы сделать, это объединить действия перетаскивания и щелчка в 1, ie. Удалите действие перетаскивания после присвоения его value действию щелчка.

Чтобы узнать, какой щелчок соответствует какому перетаскиванию, я должен взять щелчок, отметка времени которого ближе всего к timestamp перетаскивания. Я также хочу убедиться, что перетаскивание не может быть связано с кликом, если разница между отметками времени превышает 5 (и это означает, что некоторые перетаскивания могут быть не связаны, это нормально). Конечно, я не хочу, чтобы перетаскивание пользователя 1 соответствовало клику пользователя 2.

Здесь результат будет выглядеть следующим образом: 

+---------+----------+----------+---------+
|timestamp|   user_id|    action|    value|
+---------+----------+----------+---------+
|      100|         1|     click|     null|
|      101|         2|     click|     null|
|      101|         1|     click|      AAA|
|      108|         1|     click|      BBB|
|      100|         2|     click|     null|
+---------+----------+----------+---------+

Перетаскивание с помощью AAA (timestamp = 103) был связан с кликом, который произошел в 101, потому что он ближе всего к 103. Те же логы c для BBB.

Так что я хотел бы выполнить эти операции, гладко / эффективно. Пока что у меня есть что-то вроде этого: 

val window =  Window partitionBy ($"user_id") orderBy $"timestamp".asc

myDF
  .withColumn("previous_value", lag("value", 1, null) over window)
  .withColumn("previous_timestamp", lag("timestamp", 1, null) over window)
  .withColumn("next_value", lead("value", 1, null) over window)
  .withColumn("next_timestamp", lead("timestamp", 1, null) over window)

  .withColumn("value",
        when(
            $"previous_value".isNotNull and
            // If there is more than 5 sec. difference, it shouldn't be joined
            $"timestamp" - $"previous_timestamp" < 5 and
            (
                $"next_timestamp".isNull or
                $"next_timestamp" - $"timestamp" > $"timestamp" - $"previous_timestamp"
            ), $"previous_value")
        .otherwise(
            when($"next_timestamp" - $"timestamp" < 5, $"next_value")
            .otherwise(null)
        )
    )
  .filter($"action" === "click")
  .drop("previous_value")
  .drop("previous_timestamp")
  .drop("next_value")
  .drop("next_timestamp")

Но я чувствую, что это довольно неэффективно. Есть лучший способ сделать это ? (что-то, что можно сделать, не создавая 4 временных столбца ...) Можно ли, например, манипулировать строкой со смещением -1 и +1 в одном и том же выражении?

Заранее спасибо!

1 Ответ

1 голос
/ 30 марта 2020

Вот моя попытка использовать Spark- SQL вместо API DataFrame, но должна быть возможность конвертировать: 

myDF.registerTempTable("mydf")

spark.sql("""
with
 clicks_table as (select * from mydf where action='click')
,drags_table  as (select * from mydf where action='drag' )

,one_click_many_drags as (
  select
    c.timestamp as c_timestamp
    , d.timestamp as d_timestamp
    , c.user_id as c_user_id
    , d.user_id as d_user_id
    , c.action as c_action
    , d.action as d_action
    , c.value as c_value
    , d.value as d_value
  from clicks_table c
  inner join drags_table d
  on c.user_id = d.user_id
    and abs(c.timestamp - d.timestamp) <= 5 --a drag cannot be linked to a click if there timestamp difference is over 5
)
,one_click_one_drag as (
  select c_timestamp as timestamp, c_user_id as user_id, c_action as action, d_value as value
  from (
    select *, row_number() over (
      partition by d_user_id, d_timestamp --for each drag timestamp with multiple possible click timestamps, we rank the click timestamps by nearness
      order by
        abs(c_timestamp - d_timestamp) asc --prefer nearest
        , c_timestamp asc --prefer next_value if tied
    ) as rn
    from one_click_many_drags
  ) 
  where rn=1 --take only the best match for each drag timestamp
)

--now we start from the clicks_table and add in the desired drag values!

select c.timestamp, c.user_id, c.action, m.value
from clicks_table c
left join one_click_one_drag m
on c.user_id = m.user_id
and c.timestamp = m.timestamp

""")

Протестировано для получения желаемого результата.

...