Pyspark: определяемая пользователем эластичность для каждого уникального идентификатора
Я работаю над проблемой эластичности цены, где мне нужно рассчитать эластичность для каждого уникального идентификатора.
Мой фрейм данных выглядит так
| id | price | items |
| 101 | 5 | 10 |
| 101 | 10 | 15 |
| 101 | 12 | 20 |
| 102 | 1 | 1 |
| 102 | 3 | 7 |
Чтобы найти эластичность: рассмотрим пример 101, происходит 3 изменения цен, эти три изменения цен и соответствующие изменения элементов должны быть новым фреймом данных.
1) изменение цены на 5 (5 -> 10 (или 10 -> 5)) привело к изменению 5 позиций (10 -> 15 (или 15 -> 10)) поэтому соответствующая строка будет pricechange = 5, itemschange = 5
2) изменение цены на 7 (5 -> 12 (или 12 -> 5)) привело к изменению 10 позиций (10 -> 20 (или 20 -> 10)) поэтому соответствующая строка будет pricechange = 7, itemschange = 10
3) изменение цены на 2 (10 -> 12 (или 12 -> 10)) привело к изменению 5 позиций (15 -> 20 (или 20 -> 15)) поэтому соответствующая строка будет pricechange = 2, itemschange = 5
Фрейм данных будет преобразован в:
| id | pricechange | itemschange |
| 101 | 5 | 5 |
| 101 | 7 | 10 |
| 101 | 2 | 5 |
| 102 | 2 | 6 |
Простое изменение слов не решает проблему. Вы неявно подразумеваете, что цены меняются с 5 до 10 в первом примере. Как определить, что сначала идет 5, а затем 10? Почему это не может быть 10 -> 5 (искра не имеет ни малейшего представления, пока вы не скажете ему).
Здесь учитываются все комбинации. В вашем сценарии 10 идут первыми, а 5 идут следующими, абсолютная разница по-прежнему составляет 5, поэтому изменение цены по-прежнему составляет 5.
Хорошо, это имеет смысл - вы должны редактировать свой вопрос, чтобы уточнить, что вас волнует только абсолютная разница. Во всяком случае, кажется, что это можно решить с помощью простого внутреннего соединения. Присоедините фрейм данных к самому себе на id и вычислите изменения цены и изменения товара. Вы пробовали это?
Я пробовал это, но это приводит к повторяющимся строкам. Не могли бы вы сообщить мне процедуру обработки повторяющихся строк? Под повторяющимися строками я подразумеваю 10 -> 5, а также 5 -> 10.
Ответы 2
Вот подробный подход, которому вы можете следовать -
Определите схему и подготовьте данные для DF
df = spark.createDataFrame([
(101,5,10),
(101,10,15),
(101,12,20),
(102,1,1),
(102,3,7)
],'id : int, price : int, item: int')
Создайте фиктивный рейтинг столбца, чтобы сравнить каждый идентификатор со всеми другими записями с таким же идентификатором.
from pyspark.sql.window import Window
from pyspark.sql.functions import *
windowSpec = Window.partitionBy('id').orderBy('id')
rank = row_number().over(windowSpec).alias('rank')
df = df.withColumn("rank", rank)
Окончательная логика — присоединиться и фильтровать
df.alias('a').\
join(df.alias('b'),on='id').\
where('a.rank < b.rank').\
selectExpr("a.id as id","b.price - a.price as price","b.item - a.item as item").\
show()
ИМХО. Всегда лучше опубликовать то, что вы пробовали до сих пор, и с какой ошибкой/проблемой вы столкнулись вместе с вопросом. Это поможет получить быстрый и лучший ответ.
Вы можете просто выполнить внутреннее соединение DataFrame с самим собой в столбце id. Чтобы избежать дублирования записей, определите предложение where, которое требует, чтобы цена левого фрейма данных была больше, чем цена правого фрейма данных.
После объединения выберите нужные столбцы:
from pyspark.sql.functions import col
df.alias("r").join(df.alias("l"), on = "id")\
.where("l.price > r.price")\
.select(
"id",
(col("l.price") - col("r.price")).alias("pricechange"),
(col("l.item") - col("r.item")).alias("itemschange"),
).show()
#+---+-----------+-----------+
#| id|pricechange|itemschange|
#+---+-----------+-----------+
#|101| 2| 5|
#|101| 7| 10|
#|101| 5| 5|
#|102| 2| 6|
#+---+-----------+-----------+
Это будет более эффективно, чем использование Window.
искровые кадры данных неупорядочены. как вы определяете, какая строка является «первой», а какая — «следующей» строкой? Вы не можете просто полагаться на порядок в файле или на то, как он отображается - вам нужно указать его самостоятельно.