scala - Spark SQL 使用窗口 - 根据列条件从当前行之后的行中收集数据
问题描述
我有一个像这样的 spark 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) 检查 Mood 是否仅在当前行 &&Happy之后收集所有。否则，如果 Mood 是：放置空数组。item_idmood = Sadsad
2) 这必须超过window上面指定的 I。（例如，这个数据框有两个窗口->第一个是 (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如何收集_设置item_idONLY whenmood=sad？
关于如何解决这个问题的任何要点？
解决方案
在分区结束后，我无法在下一行和下一行之间给出行。所以我使用了当前行和无限跟随，然后使用 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 |[] |[] |
+-------+-------+-----+----+------------+------------------------------+---------+
编辑1：
正如 OP 所提到的，“”可以替换为nulldf3 本身将是最终结果。因此 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>
推荐阅读
- node.js - node.js catch 块在等待时未按预期触发
- css - 在同一行中对齐表单输入
- docker - 是否可以运行一个接受私有图像的推送请求和来自 dockerhub 的代理图像的 habor 实例？
- batch-file - 如何根据 ID 分隔 2 个 TXT 文件
- javascript - Javascript录音机在android Webview中不起作用
- javascript - 如何同步进行ajax验证
- flutter - Flutter 嵌套列表大小
- gcc - movsbl near ret 对性能有好处吗？
- php - 扩展 Sonata 的 PageAdmin 时找不到自定义方法
- python - 如何在边界框区域内获取热图的平均值