Создание попарных кумулятивных статистических данных по строкам данных

Question

У меня есть таблица с 3 столбцами: дата, идентификатор и ключ. Я надеюсь найти эффективный способ суммирования парных экземпляров ключей в идентификаторе, а затем объединить их с итоговыми значениями по другим идентификаторам. В основном построение списка временных переходов. Например:

Введите:

╔══════════╦════╦═════╗
║   Date   ║ ID ║ Key ║
╠══════════╬════╬═════╣
║ 1/1/2018 ║ A  ║ XY  ║
║ 1/2/2018 ║ A  ║ GT  ║
║ 1/6/2018 ║ A  ║ WE  ║
║ 1/9/2018 ║ A  ║ PO  ║
║ 1/2/2018 ║ B  ║ XY  ║
║ 1/4/2018 ║ B  ║ GT  ║
╚══════════╩════╩═════╝

Выход:

╔══════════╦═══════════╦═══════╗
║ FirstKey ║ SecondKey ║ Count ║
╠══════════╬═══════════╬═══════╣
║    XY    ║    GT     ║   2   ║
║    GT    ║    WE     ║   1   ║
║    WE    ║    PO     ║   1   ║
╚══════════╩═══════════╩═══════╝

Было бы тривиально просто упорядочить по идентификатору, затем Дата затем просто перебрать и построить счетчик, но я надеялся, что один из вас, гуру, сможет помочь мне структурировать его, чтобы он был более распараллеленным / эффективным.

По сути, так как упорядочено по дате, я пытаюсь зафиксировать количество переходов во времени между клавишами. Таким образом, для ID = A у нас есть XY, тогда у нас есть GT (поэтому увеличьте XY-> GT на 1). Затем у нас есть GT, за которым следует WE (поэтому увеличим GT-> PO на 1).

Работа над искрой с помощью scala / python.

Answer 1

Вот решение в Scala, использующее lag(Key, 1) для объединения предыдущих / текущих ключей для подсчета пары ключей:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
import spark.implicits._

val df = Seq(
  ("1/1/2018", "A", "XY"),
  ("1/2/2018", "A", "GT"),
  ("1/6/2018", "A", "WE"),
  ("1/9/2018", "A", "PO"),
  ("1/2/2018", "B", "XY"),
  ("1/4/2018", "B", "GT")
).toDF("Date", "ID", "Key")

val win = Window.partitionBy("ID").orderBy("Date", "Key")

df.
  withColumn("Date", to_date($"Date", "M/d/yyyy")).
  withColumn("FirstKey", lag($"Key", 1).over(win)).
  groupBy($"FirstKey", $"Key".as("SecondKey")).agg(count("*").as("Count")).
  where($"firstKey".isNotNull).
  show
// +--------+---------+-----+
// |FirstKey|SecondKey|Count|
// +--------+---------+-----+
// |      WE|       PO|    1|
// |      GT|       WE|    1|
// |      XY|       GT|    2|
// +--------+---------+-----+

Обратите внимание, что преобразование to_date предназначено для обеспечения правильного хронологического упорядочения.

Answer 2

Вы можете добавить новый столбец, который показывает следующий Key для каждого ID, упорядоченного по Date, используя pyspark.sql.functions.lead. Затем сгруппируйте по FirstKey и SecondKey и count:

from pyspark.sql import Window
from pyspark.sql.functions import col, lead

df.withColumn("SecondKey", lead("Key").over(Window.partitionBy("ID").orderBy("Date")))\
    .where(col("SecondKey").isNotNull())\
    .groupBy(col("Key").alias("FirstKey"), "SecondKey")\
    .count()\
    .show()
#+--------+---------+-----+
#|FirstKey|SecondKey|count|
#+--------+---------+-----+
#|      WE|       PO|    1|
#|      GT|       WE|    1|
#|      XY|       GT|    2|
#+--------+---------+-----+

Предполагается, что столбец Date представляет собой DateType, поэтому его можно упорядочить соответствующим образом. Если это строка, вам нужно преобразовать ее в дату , иначе порядок будет лексикографическим.

Answer 3

Вот потенциальное решение, требующее всего 3 строки:

import pandas as pd

df = pd.DataFrame({'Date': ['1/1/2018', '1/2/2018', '1/6/2018', '1/9/2018', '1/2/2018', '1/4/2018'], 'ID': ['A', 'A', 'A', 'A', 'B', 'B'], 'Key': ['XY', 'GT', 'WE', 'PO', 'XY', 'GT']})
print(df)


       Date ID Key
0  1/1/2018  A  XY
1  1/2/2018  A  GT
2  1/6/2018  A  WE
3  1/9/2018  A  PO
4  1/2/2018  B  XY
5  1/4/2018  B  GT

df['key_lag'] = df.Key.shift(-1)
df['key_pairs'] = df.Key + ' ' + df.key_lag
print(df.groupby('key_pairs').size())


key_pairs
GT WE    1
PO XY    1
WE PO    1
XY GT    2
dtype: int64