pyspark - 在滑动窗口上应用 UDF 时出现索引错误答案

【问题标题】：pyspark - Index Error when applying UDF over Sliding Windowpyspark - 在滑动窗口上应用 UDF 时出现索引错误
【发布时间】：2021-12-16 05:17:21
【问题描述】：

我正在使用包含用户、日期和位置的 pyspark 数据框。我的目标是实现一个 3 天 [-1 天，1 天] 的滑动窗口并计算窗口内最常见的位置。

+---+-------------+----+------------+---------+
| ID|         date| loc| GOAL_Window| GOAL_Loc|
+---+-------------+----+------------+---------+
|ID1|   2017-07-01|  L1|     [L1,L1]|       L1|
|ID1|   2017-07-02|  L1|  [L1,L1,L5]|       L1|
|ID1|   2017-07-03|  L5|  [L1,L5,L1]|       L1|
|ID1|   2017-07-04|  L1|  [L5,L1,L5]|       L5|
|ID1|   2017-07-05|  L5|  [L1,L5,L5]|       L5|
|ID1|   2017-07-06|  L5|     [L5,L5]|       L5|
|ID1|   2017-07-08|  L5|        [L5]|       L5|
|ID2|   2017-07-01|  L0|     [L0,L0]|       L0|
|ID2|   2017-07-02|  L0|        [L0]|       L0|
+---+-------------+----+------------+---------+

对于每一行，我需要：

选择滑动窗口。适用于：

days = lambda i: i*86400
w = Window.partitionBy('id').orderBy(F.col('date').cast('timestamp').cast('long')\
     .rangeBetween(-days(-1),days(1))

发件人：Spark Window Functions - rangeBetween dates

创建一个应用于窗口并计算的 UDF：
2.a.对于窗口的每个元素，计算该位置出现的次数
2.b。按降序排列计数系列
2.c。选择计数最多的位置 ID
我应用它的灵感来自：Improve Pandas UDF in Pyspark

import pandas as pd
from typing import List
from pyspark.sql.types import StringType

@F.udf(StringType)
def pd_ctfirst(dt: List[str]) -> str:
   df = pd.DataFrame({'loc':loc})
   df = df.reset_index().groupby('loc').count()
   return str(df.reset_index().sort_values('index', ascending = False)['loc'].values[0])

df_ = df_.withColumn('GOAL_Window', F.collect_list(F.col('loc')).over(w))\         
         .withColumn('GOAL_Loc', pd_ctfirst(F.collect_list(F.col('loc')).over(w)))

检查数据框（例如：df_.take()）时，结果看起来正确，但我无法保存或订购。这样做时，我得到错误：

IndexError: index 0 is out of bounds for axis 0 with size 0

关于我为什么会收到错误以及如何将计算应用于每个窗口幻灯片的任何建议？提前致谢！

【问题讨论】：

标签： python pyspark window user-defined-functions sliding-window

【解决方案1】：

如果有人发现类似情况，我设法通过添加对 df 大小的检查来克服这个问题：

@F.udf(StringType)
def pd_ctfirst(dt: List[str]) -> str:
   df = pd.DataFrame({'loc':loc})
   if df.size != 0:
     df = df.reset_index().groupby('loc').count()
     return str(df.reset_index().sort_values('index', ascending = False)['loc'].values[0])
   else:
     return None

【讨论】：