如何计算scalaDataframe中列的特定值的记录更改

mwkjh3gx 于 2021-05-27 发布在 Spark

关注(0)|答案(1)|浏览(529)

在Dataframe中，列的输入如下所示：

| id|  priority|         status|       datetime|data_as_of_Date|Amount|open_close|
    |  1|Unassigned|          Fixed| 10/8/2019 0:00| 2/12/2020 0:00|    40|    Closed|
    |  1|Unassigned|            New|2/12/2019 11:00| 2/12/2020 0:00|    20|      Open|
    |  1|Unassigned|Fix in progress|9/12/2019 11:00| 2/12/2020 0:00|    90|      Open|
    |  3|  Critical|        Removed|5/17/2019 12:00| 2/12/2020 0:00|    33|    Closed|
    |  3|Unassigned|Fix in progress|5/26/2019 10:00| 2/12/2020 0:00|    30|      Open|
    |  3|  Critical|            New|  5/8/2019 3:00| 2/12/2020 0:00|    34|      Open|
    |  3|Unassigned|          Fixed| 7/29/2019 7:00| 2/12/2020 0:00|    29|    Closed|

我该如何计算 open_close 每个公司的专栏都换了？

scala apache-spark

来源：https://stackoverflow.com/questions/61687176/how-to-count-record-changes-for-a-particular-value-of-a-column-in-a-scala-datafr

1条答案

按热度按时间

91zkwejq1#

可以使用窗口函数使用日期列添加行号。然后使用lag函数创建一个向下移动一个位置的新列，如果open\u close值与前一个不同，则放入“1”，否则放入“0”。最后，按公司id分组并将更改的总和标记为1。

val df2 = df.withColumn("row_num",row_number.over(Window.orderBy('datetime).partitionBy('id)))
val df3 = df2.select('*,lag('open_close, 1, 0).over(Window.orderBy('row_num).partitionBy('id)).as("lag"))
val df4 = df3.select('*,when('open_close === 'lag || 'lag === 0 , 0).otherwise(1).as("change"))
df4.groupBy('id).agg(sum('change)).show()

+---+-----------+
| id|sum(change)|
+---+-----------+
|  1|          1|
|  3|          2|
+---+-----------+

赞(0）回复(0）举报 2021-05-27

我来回答

如何计算scalaDataframe中列的特定值的记录更改

1条答案

相关问题

热门标签

最新问答