使用窗口根据列条件从当前行后的行中收集数据。
使用窗口根据列条件从当前行后的行中收集数据。
提问于 2018-11-29 17:43:05
我有一个火花DataFrame (在Scala中)，如下所示：
+---------+-------------+------+---------+------------+
| user_id| item_id| mood| time|sessionBegin|
+---------+-------------+------+---------+------------+
| 1| A| Happy| 0| 0|
| 1| B| Happy| 1| 0|
| 1| C| Happy| 3| 0|
| 1| D| Happy| 5| 0|
| 1| C| Happy| 6| 0|
| 1| D| Sad| 6| 0|
| 1| C| Sad| 10| 0|
| 1| A| Happy| 28| 0|
| 1| E| Happy| 35| 0|
| 1| E| Sad| 60| 0|
| 2| F| Happy| 6| 6|
| 2| E| Happy| 17| 6|
| 2| D| Happy| 20| 6|
| 2| D| Sad| 21| 6|
| 2| E| Happy| 27| 6|
| 2| G| Happy| 37| 6|
| 2| H| Happy| 39| 6|
| 2| G| Sad| 45| 6|
+---------+-------------+------+---------+------------+我在列(user_id，sessionBegin)上定义了一个窗口，并按时间排序。
val window = Window.partitionBy("user_id","sessionBegin").orderBy("time")
现在，我想添加一个列result，它：
1)检查心情是否为Happy，然后只在当前行&& mood = Sad之后收集所有item_id。否则，如果心情是sad：放入空数组。
2)这必须超出上面指定的window i。(例如)。此数据有两个->窗口，一个是is (user_id =1，sessionBegin = 0)，另一个是(user_id = 2，sessionBegin = 6)。
因此，最终的DF将是：
+---------+-------------+------+---------+------------+---------+
| user_id| item_id| mood| time|sessionBegin| result|
+---------+-------------+------+---------+------------+---------+
| 1| A| Happy| 0| 0| [D,C,E]|
| 1| B| Happy| 1| 0| [D,C,E]|
| 1| C| Happy| 3| 0| [D,C,E]|
| 1| D| Happy| 5| 0| [D,C,E]|
| 1| C| Happy| 6| 0| [D,C,E]|
| 1| D| Sad| 6| 0| []|
| 1| C| Sad| 10| 0| []|
| 1| A| Happy| 28| 0| [E]|
| 1| E| Happy| 35| 0| [E]|
| 1| E| Sad| 60| 0| []|
| 2| F| Happy| 6| 6| [D,G]|
| 2| E| Happy| 17| 6| [D,G]|
| 2| D| Happy| 20| 6| [D,G]|
| 2| D| Sad| 21| 6| []|
| 2| E| Happy| 27| 6| [G]|
| 2| G| Happy| 37| 6| [G]|
| 2| H| Happy| 39| 6| [G]|
| 2| G| Sad| 45| 6| []|
+---------+-------------+------+---------+------------+---------+我在窗口上使用了使用collect_set的when..otherwise方法，但是我找不到两件事：
- 如何仅考虑当前行之后的行
- 对于使用
mood=Happy的所有行，如何仅当mood=sad时才能对item_id进行collect_set？
有什么办法解决这个问题吗？
回答 1
Stack Overflow用户
回答已采纳
发布于 2018-11-30 12:45:25
我无法在分区结束后的下一行和下一行之间给出行。因此，我使用了当前行和无界跟随，然后使用udf删除了第一个Array元素。我使用了所有的- spark.sql，udf和df操作。看看这个
val df = Seq((1,"A","Happy","0","0"),(1,"B","Happy","1","0"),(1,"C","Happy","3","0"),(1,"D","Happy","5","0"),(1,"C","Happy","6","0"),(1,"D","Sad","6","0"),(1,"C","Sad","10","0"),(1,"A","Happy","28","0"),(1,"E","Happy","35","0"),(1,"E","Sad","60","0"),(2,"F","Happy","6","6"),(2,"E","Happy","17","6"),(2,"D","Happy","20","6"),(2,"D","Sad","21","6"),(2,"E","Happy","27","6"),(2,"G","Happy","37","6"),(2,"H","Happy","39","6"),(2,"G","Sad","45","6")).toDF("user_id","item_id","mood","time","sessionBegin")
val df2 = df.withColumn("time", 'time.cast("int"))
df2.createOrReplaceTempView("user")
val df3 = spark.sql(
"""
select user_id, item_id, mood, time, sessionBegin,
case when mood='Happy' then
collect_list(case when mood='Happy' then ' ' when mood='Sad' then item_id end) over(partition by user_id order by time rows between current row and unbounded following )
when mood='Sad' then array()
end as result from user
""")
def sliceResult(x:Seq[String]):Seq[String]={
val y = x.drop(1).filter( _ != " ")
y.toSet.toSeq
}
val udf_sliceResult = udf ( sliceResult(_:Seq[String]):Seq[String] )
df3.withColumn("result1", udf_sliceResult('result) ).show(false)结果：
+-------+-------+-----+----+------------+------------------------------+---------+
|user_id|item_id|mood |time|sessionBegin|result |result1 |
+-------+-------+-----+----+------------+------------------------------+---------+
|1 |A |Happy|0 |0 |[ , , , , , D, C, , , E]|[D, C, E]|
|1 |B |Happy|1 |0 |[ , , , , D, C, , , E] |[D, C, E]|
|1 |C |Happy|3 |0 |[ , , , D, C, , , E] |[D, C, E]|
|1 |D |Happy|5 |0 |[ , , D, C, , , E] |[D, C, E]|
|1 |C |Happy|6 |0 |[ , D, C, , , E] |[D, C, E]|
|1 |D |Sad |6 |0 |[] |[] |
|1 |C |Sad |10 |0 |[] |[] |
|1 |A |Happy|28 |0 |[ , , E] |[E] |
|1 |E |Happy|35 |0 |[ , E] |[E] |
|1 |E |Sad |60 |0 |[] |[] |
|2 |F |Happy|6 |6 |[ , , , D, , , , G] |[D, G] |
|2 |E |Happy|17 |6 |[ , , D, , , , G] |[D, G] |
|2 |D |Happy|20 |6 |[ , D, , , , G] |[D, G] |
|2 |D |Sad |21 |6 |[] |[] |
|2 |E |Happy|27 |6 |[ , , , G] |[G] |
|2 |G |Happy|37 |6 |[ , , G] |[G] |
|2 |H |Happy|39 |6 |[ , G] |[G] |
|2 |G |Sad |45 |6 |[] |[] |
+-------+-------+-----+----+------------+------------------------------+---------+EDIT1:
正如OP所提到的，“”可以用null替换，而df3本身将是最终的结果。因此，可以避免udf()
scala> :paste
// Entering paste mode (ctrl-D to finish)
val df3 = spark.sql(
"""
select user_id, item_id, mood, time, sessionBegin,
case when mood='Happy' then
collect_list(case when mood='Happy' then null when mood='Sad' then item_id end) over(partition by user_id order by time rows between current row and unbounded following )
when mood='Sad' then array()
end as result from user
""")
// Exiting paste mode, now interpreting.
df3: org.apache.spark.sql.DataFrame = [user_id: int, item_id: string ... 4 more fields]
scala> df3.show(false)
+-------+-------+-----+----+------------+---------+
|user_id|item_id|mood |time|sessionBegin|result |
+-------+-------+-----+----+------------+---------+
|1 |A |Happy|0 |0 |[D, C, E]|
|1 |B |Happy|1 |0 |[D, C, E]|
|1 |C |Happy|3 |0 |[D, C, E]|
|1 |D |Happy|5 |0 |[D, C, E]|
|1 |C |Happy|6 |0 |[D, C, E]|
|1 |D |Sad |6 |0 |[] |
|1 |C |Sad |10 |0 |[] |
|1 |A |Happy|28 |0 |[E] |
|1 |E |Happy|35 |0 |[E] |
|1 |E |Sad |60 |0 |[] |
|2 |F |Happy|6 |6 |[D, G] |
|2 |E |Happy|17 |6 |[D, G] |
|2 |D |Happy|20 |6 |[D, G] |
|2 |D |Sad |21 |6 |[] |
|2 |E |Happy|27 |6 |[G] |
|2 |G |Happy|37 |6 |[G] |
|2 |H |Happy|39 |6 |[G] |
|2 |G |Sad |45 |6 |[] |
+-------+-------+-----+----+------------+---------+
scala>页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/53544757
复制相关文章
相似问题
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号