pyspark: скользящее среднее с заполнением нулями данных временных рядов - PullRequest
Новая
       48

pyspark: скользящее среднее с заполнением нулями данных временных рядов

0 голосов
/ 09 июля 2019

У меня есть набор данных, который содержит данные о распродаже за несколько недель.Я хочу рассчитать скользящее среднее, например, за 3 недели, но с учетом недель, в которых нет продаж.

Давайте рассмотрим следующие данные: 

|------|-------|
|wk_id |sellout|
|------|-------|
|201801|    1.0|
|201802|    5.0|
|201803|    3.0|
|201805|    1.0|
|201806|    5.0|
|------|-------|

Мой ожидаемый результат: 

|------|-------|-------------|
|wk_id |sellout|moving_avg_3w|
|------|-------|-------------|
|201801|    1.0|0.333        | <- (0+0+1)/3
|201802|    5.0|2.000        | <- (0+1+5)/3
|201803|    3.0|3.000        | <- (1+5+3)/3
|201805|    1.0|1.333        | <- (3+0+1)/3
|201806|    5.0|2.000        | <- (5+1+0)/3
|------|-------|-------------|

Моим наивным решением было бы заполнить недостающие недели 0, а затем использовать подход, который был представлен здесь: pyspark: скользящее среднее с использованием данных временных рядов

Но если у вас есть огромное количество данных, это не самый эффективный подход.У кого-нибудь есть лучшее решение?

Этот вопрос о PySpark

1 Ответ

2 голосов
/ 10 июля 2019

Таким образом, вы можете фактически использовать метод в ссылке, которую вы разместили с rangeBetween вместо window, как только вы измените 'wk_id' на unix_timestamp, чтобы получить достаточное пространство между неделями. 

import pyspark.sql.functions as F
from pyspark.sql.window import Window

# create the df: some wk_id are different to see it works when you change year as well
df = spark.createDataFrame( [ (201801, 1.0), (201802, 5.0), (201804,3.0), 
                              (201851, 3.0), (201852, 1.0), (201901,5.0)], 
                            ['wk_id','sellout'])
# nb_wk you want to roll over
nb_wk = 3

# function to calculate the number of seconds from the number of weeks
wk_to_sec = lambda i: i * 7 * 86400

# create the window of nb_wk
w = Window().orderBy(F.col("sec")).rangeBetween(-wk_to_sec(nb_wk-1), 0)

# add the columns of the number of seconds then the moving average by a sum divide by nb_wk
# the method mean does not work here as there are missing weeks
df = df.withColumn( 'sec', F.unix_timestamp(F.col('wk_id').cast('string'), format="YYYYww"))\
       .withColumn( 'moving_avg_{}w'.format(nb_wk), F.sum('sellout').over(w)/nb_wk)

df.show()
+------+-------+----------+------------------+
| wk_id|sellout|       sec|     moving_avg_3w|
+------+-------+----------+------------------+
|201801|    1.0|1514696400|0.3333333333333333|
|201802|    5.0|1515301200|               2.0|
|201804|    3.0|1516510800|2.6666666666666665| # here it is (5+0+3)/3
|201851|    3.0|1544936400|               1.0|
|201852|    1.0|1545541200|1.3333333333333333|
|201901|    5.0|1546146000|               3.0|  # here it is (3+1+5)/3
+------+-------+----------+------------------+

Вы можете удалить столбец 'sec' после или, если вы не хотите создавать этот столбец, вы можете сделать все сразу: 

# create the window of nb_wk with unix_timestamp directly in it
w = Window().orderBy(F.unix_timestamp(F.col('wk_id').cast('string'), format="YYYYww"))
            .rangeBetween(-wk_to_sec(nb_wk-1), 0)
df = df.withColumn( 'moving_avg_{}w'.format(nb_wk), F.sum('sellout').over(w)/nb_wk)

РЕДАКТИРОВАТЬ: для перемещения стандартного отклонения, я думаю, вы можете сделать это так, но не уверены в производительности: 

df = df.withColumn('std', F.sqrt( (F.sum( (F.col('sellout') - F.last('roll_mean_3w').over(w))**2).over(w) 
                                   + (nb_wk - F.count('sellout').over(w))*F.last('roll_mean_3w').over(w)**2)
                                   /nb_wk))
...