在 spark 数据框中连续应用更改答案

【问题标题】：Apply changes consecutively in a spark dataframe在 spark 数据框中连续应用更改
【发布时间】：2019-07-02 14:00:13
【问题描述】：

我有一个初始状态名为 init 的数据框。我有一个具有相同架构的数据框，其中每行更新一个数据框初始化字段，其他字段为 Null。如何连续应用更改重建每条记录？为了更清楚，让我们举个例子：

listOfTuples = [(101, "Status_0", '2019','value_col_4',0)]
init = spark.createDataFrame(listOfTuples , ["id", "status", "year","col_4","ord"])

#initial state
>>> init.show()
+---+--------+----+-----------+---+
| id|  status|year|      col_4|ord|
+---+--------+----+-----------+---+
|  1|Status_0|2019|value_col_4|  0|
+---+--------+----+-----------+---+

#dataframe with changes
schema = StructType([StructField('id', StringType(), True),
                     StructField('status', StringType(), True),
                     StructField('year', StringType(), True),
                     StructField('col_4', StringType(), True),
                     StructField('ord', IntegerType(), True)])

listOfTuples = [(1, "Status_A", None, None,1),
                (1, "Status_B", None, None,2),
                (1, None, None, "new_val", 3),
                (1, "Status_C", None, None,4)]


changes = spark.createDataFrame(listOfTuples ,  schema)
>>> changes.show()
+---+--------+----+-------+---+
| id|  status|year|  col_4|ord|
+---+--------+----+-------+---+
|  1|Status_A|null|   null|  1|
|  1|Status_B|null|   null|  2|
|  1|    null|null|new_val|  3|
|  1|Status_C|null|   null|  4|
+---+--------+----+-------+---+

我希望更改以 ord 列的顺序连续应用于最终数据帧，并为数据帧 init 中的值设置基线。所以我希望我的最终数据框是这样的：

>>> final.show()
+---+--------+----+--------------+
| id|  status|year|  col_4       |
+---+--------+----+--------------+
|  1|Status_0|2019|  value_col_4 |
|  1|Status_A|2019|  value_col_4 |
|  1|Status_B|2019|  value_col_4 |
|  1|Status_B|2019|  new_val     |
|  1|Status_C|2019|  new_val     |
+---+--------+----+--------------+

我正在考虑合并两个按 ord 列排序的数据框，然后以某种方式在下面传播更改。有谁知道如何做到这一点？

【问题讨论】：

是否可以使用窗口函数来做到这一点，所以按id 分区，按ord 排序，然后为每一列与前一行做一个coalesce。不过我不确定这是否可行。
只有当当前行没有更新值（空列）时，我才需要取上一行的值 - 我不确定这将如何与合并一起工作。
coalesce 接受 2 个参数。如果第一个为空，则返回第二个。所以说你做df.withColumn(c, F.coalesce(F.col(c), F.lag(c).over(win))，如果有变化，它将用更新替换前一行的值。我唯一不确定的是更新是累积的还是你只会得到空值。

标签： apache-spark dataframe pyspark

【解决方案1】：

这是 Scala 代码，但我希望这会有所帮助。最后，您可以删除或重命名列。解决方案是执行union，然后在unboundedpreceding 行的框架中为所有3 列获取org.apache.spark.sql.functions.last 非空值到currentrow。

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.expressions.WindowSpec
import org.apache.spark.sql.functions._

scala> initial.show
+---+--------+----+-----------+---+
| id|  status|year|      col_4|ord|
+---+--------+----+-----------+---+
|  1|Status_0|2019|value_col_4|  0|
+---+--------+----+-----------+---+

scala> changes.show
+---+--------+----+-------+---+
| id|  status|year|  col_4|ord|
+---+--------+----+-------+---+
|  1|Status_A|null|   null|  1|
|  1|Status_B|null|   null|  2|
|  1|    null|null|new_val|  3|
|  1|Status_C|null|   null|  4|
+---+--------+----+-------+---+


scala> val inter = initial.union(changes)
inter: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: string, status: string ... 3 more fields]

scala> inter.show
+---+--------+----+-----------+---+
| id|  status|year|      col_4|ord|
+---+--------+----+-----------+---+
|  1|Status_0|2019|value_col_4|  0|
|  1|Status_A|null|       null|  1|
|  1|Status_B|null|       null|  2|
|  1|    null|null|    new_val|  3|
|  1|Status_C|null|       null|  4|
+---+--------+----+-----------+---+


scala> val overColumns = Window.partitionBy("id").orderBy("ord").rowsBetween(Window.unboundedPreceding, Window.currentRow)
overColumns: org.apache.spark.sql.expressions.WindowSpec = org.apache.spark.sql.expressions.WindowSpec@70f4b378

scala> val output = inter.withColumn("newstatus", 
  last("status", true).over(overColumns)).withColumn("newyear",
  last("year", true).over(overColumns)).withColumn("newcol_4", 
  last("col_4", true).over(overColumns))
output: org.apache.spark.sql.DataFrame = [id: string, status: string ... 6 more fields]

scala> output.show(false)
+---+--------+----+-----------+---+---------+-------+-----------+
|id |status  |year|col_4      |ord|newstatus|newyear|newcol_4   |
+---+--------+----+-----------+---+---------+-------+-----------+
|1  |Status_0|2019|value_col_4|0  |Status_0 |2019   |value_col_4|
|1  |Status_A|null|null       |1  |Status_A |2019   |value_col_4|
|1  |Status_B|null|null       |2  |Status_B |2019   |value_col_4|
|1  |null    |null|new_val    |3  |Status_B |2019   |new_val    |
|1  |Status_C|null|null       |4  |Status_C |2019   |new_val    |
+---+--------+----+-----------+---+---------+-------+-----------+

【讨论】：

【解决方案2】：

在 python 中使用来自@C.S.Reddy Gadipally 的代码

import pyspark.sql.functions as f
from pyspark.sql.window import Window

f = init.union(changes)

w = Window.partitionBy(f['id']).orderBy(f['ord'])

for c in f.columns[1:]:
    f = f.withColumn(c,func.last(c,True).over(w))

【讨论】：