如何解决Pyspark 跨行保留值？
我有一个问题，可以使用逐行 SAS 方法自然解决，但我一直在使用 Pyspark。我有一个按时间排序的人的事件数据集，例如：
test_df = pd.DataFrame({'event_list':[["H"],["H"],["H","F"],["F"],["W"],["W"]],'time_order':[1,2,3,4,5,6,7,8],'person':[1,1,1]})
test_df = spark.createDataFrame(test_df)
test_df.show()
+----------+----------+------+
|event_list|time_order|person|
+----------+----------+------+
| [H]| 1| 1|
| [H]| 2| 1|
| [H,F]| 3| 1|
| [F]| 4| 1|
| [F]| 5| 1|
| [H]| 6| 1|
| [W]| 7| 1|
| [W]| 8| 1|
+----------+----------+------+
我想将这些事件分成几集，其中初始事件之后的所有事件都是初始事件列表的一部分。因此，在我的 test_df 中，我希望有 3 集：
+----------+----------+------+-------+
|event_list|time_order|person|episode|
+----------+----------+------+-------+
| [H]| 1| 1| 1|
| [H]| 2| 1| 1|
| [H,F]| 3| 1| 2|
| [F]| 4| 1| 2|
| [F]| 5| 1| 2|
| [H]| 6| 1| 2|
| [W]| 7| 1| 3|
| [W]| 8| 1| 3|
+----------+----------+------+-------+
在 SAS 中，我将保留 event_list 的先前行的值，如果当前 event_list 包含在先前 event_list 中，我将保留当前 event_list 值而不是先前 event_list。例如。我保留的值是 [null,[“W”]]。然后我可以通过跟踪保留值的变化来生成剧集。
在 Pyspark 中，我不确定如何跨行操作按顺序保留信息......这甚至可能吗？我尝试使用窗口函数（按 person 分区并按 time_order 排序）失败了。如何在 Pyspark 中解决这个问题？
解决方法
如果您使用的是 spark 版本 >= 2.4，请在窗口上方的 event_list 列上使用 collect_list，flatten 它们，使用 array_distinct 删除重复项，最后使用size 来计算一段时间内有多少不同的事件。应该是这样的：
from pyspark.sql.functions import col,collect_list,flatten,array_distinct,size
from pyspark.sql.window import Window
w = Window.partitionBy('person').orderBy('time_order').rowsBetween(Window.unboundedPreceding,0)
test_df = test_df.withColumn('episode',size(array_distinct(flatten(collect_list(col('event_list')).over(w)))))
test_df.show()
+----------+----------+------+-------+
|event_list|time_order|person|episode|
+----------+----------+------+-------+
| [H]| 1| 1| 1|
| [H]| 2| 1| 1|
| [H,F]| 3| 1| 2|
| [F]| 4| 1| 2|
| [F]| 5| 1| 2|
| [H]| 6| 1| 2|
| [W]| 7| 1| 3|
| [W]| 8| 1| 3|
+----------+----------+------+-------+
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
