使用特定条件更新数据框

Posted

技术标签:

【中文标题】使用特定条件更新数据框【英文标题】:Update the dataframe with specific conditions 【发布时间】:2019-09-03 04:26:47 【问题描述】:

我的数据框如下:

RankNumber  Value   Dept Number
  5          200    5
  4          200    5
  3          205    5
  2          198    5
  1          197    5
  5          200    6
  4          202    6
  3          205    6
  2          198    6
  1          194    6

我想从数据框中更新值列中的一些单元格。如果当前的“值”大于先前的值,那么它应该更新为先前的值。如果“值”与之前的值相同或小于之前的值,那么它应该跳过。它已按部门编号分组。

我正在尝试在 pyspark 上执行此操作,但找不到实现该操作的方法。有人可以帮忙吗?

数据框的预期结果如下:

RankNumber  Value  Dept Number
  5         200     5
  4         200     5
  3         200     5 (record updated)
  2         198     5
  1         197     5
  5         200     6
  4         200     6 (record updated)
  3         200     6 (record updated)
  2         198     6
  1         194     6


【问题讨论】:

【参考方案1】:

我相信您的第 8 行将更新为“3 202 6 (record updated)”而不是 '3 200 6 (record updated)'。因为它之前的值是 '202' 并且当前值 '205' 大于之前的 '202'。

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

w=Window.partitionBy("DeptNumber").orderBy(desc("RankNumber"))
df = df.withColumn('previous_value',F.coalesce(F.lag(df['value'],1).over(w),df['value']))

如果 Value 大于之前的值,下面的代码将获取之前的值。

newdf = df.select(df.RankNumber,df.DeptNumber,df.Value,df.previous_value,when( df.Value<=df.previous_value, df.Value).otherwise(df.previous_value).alias('newValue'))

>>> newdf.show()
+----------+----------+-----+--------------+--------+
|RankNumber|DeptNumber|Value|previous_value|newValue|
+----------+----------+-----+--------------+--------+
|         5|         6|  200|           200|     200|
|         4|         6|  202|           200|     200|
|         3|         6|  205|           202|     202|
|         2|         6|  198|           205|     198|
|         1|         6|  194|           198|     194|
|         5|         5|  200|           200|     200|
|         4|         5|  200|           200|     200|
|         3|         5|  205|           200|     200|
|         2|         5|  198|           205|     198|
|         1|         5|  197|           198|     197|
+----------+----------+-----+--------------+--------+

以下代码将获取先前值的最小值作为新值。

from pyspark.sql.window import Window
import pyspark.sql.functions as F
from pyspark.sql.functions import desc,when,lit

w=Window.partitionBy("DeptNumber").orderBy(desc("RankNumber"))

df = df.withColumn('previous_value',F.coalesce(F.lag(df['value'],1).over(w),df['value']))

newdf = df.select(df.RankNumber,df.DeptNumber,df.Value,df.previous_value,when( df.Value<=df.previous_value, df.Value) \
                        .when(F.lag(df['previous_value'],1).over(w)<=df.previous_value, F.first(df.previous_value).over(w)) \
                        .otherwise(df.previous_value).alias('newValue'))


>>> newdf.show()
+----------+----------+-----+--------------+--------+
|RankNumber|DeptNumber|Value|previous_value|newValue|
+----------+----------+-----+--------------+--------+
|         5|         6|  200|           200|     200|
|         4|         6|  202|           200|     200|
|         3|         6|  205|           202|     200|
|         2|         6|  198|           205|     198|
|         1|         6|  194|           198|     194|
|         5|         5|  200|           200|     200|
|         4|         5|  200|           200|     200|
|         3|         5|  205|           200|     200|
|         2|         5|  198|           205|     198|
|         1|         5|  197|           198|     197|
+----------+----------+-----+--------------+--------+

如果您正在寻找一个刚好高于该组先前值的最低值,那么您需要更改这样的代码。

newdf = df.select(df.RankNumber,df.DeptNumber,df.Value,df.previous_value,when( df.Value<=df.previous_value, df.Value) \
                        .when(F.lag(df['previous_value'],1).over(w)<=df.previous_value, F.lag(df['previous_value'],1).over(w)) \
                        .otherwise(df.previous_value).alias('newValue'))

这将导致:

>>> newdf.show()
+----------+----------+-----+--------------+--------+
|RankNumber|DeptNumber|Value|previous_value|newValue|
+----------+----------+-----+--------------+--------+
|         5|     Dept2|  100|           100|     100|
|         4|     Dept2|  102|           100|     100|
|         3|     Dept2|  105|           102|     100|
|         2|     Dept2|  198|           105|     102|
|         1|     Dept2|  194|           198|     194|
|         5|     Dept1|  200|           200|     200|
|         4|     Dept1|  202|           200|     200|
|         3|     Dept1|  205|           202|     200|
|         2|     Dept1|  198|           205|     198|
|         1|     Dept1|  194|           198|     194|
+----------+----------+-----+--------------+--------+

更新: 现在创建一个新的数据框,如下面的评论部分所述:

listOfTuples = [(5, 200, "Dept1"), (4, 202, "Dept1"), (3, 205, "Dept1"), (2, 198, "Dept1"), (1, 194, "Dept1") , (5, 100, "Dept2"), (4, 102, "Dept2"), (3, 105, "Dept2"), (2, 198, "Dept2"), (1, 194, "Dept2") ]

df = spark.createDataFrame(listOfTuples , ["RankNumber", "Value", "DeptNumber"])


>>> df.show()
+----------+-----+----------+
|RankNumber|Value|DeptNumber|
+----------+-----+----------+
|         5|  200|     Dept1|
|         4|  202|     Dept1|
|         3|  205|     Dept1|
|         2|  198|     Dept1|
|         1|  194|     Dept1|
|         5|  100|     Dept2|
|         4|  102|     Dept2|
|         3|  105|     Dept2|
|         2|  198|     Dept2|
|         1|  194|     Dept2|
+----------+-----+----------+

我相信您的意图是查看当前行和前一行之间的范围,并在满足第一个条件时选择最低值。即:值大于之前的值。

w1=Window.partitionBy("DeptNumber").orderBy(desc("RankNumber"))
w2=Window.partitionBy("DeptNumber").orderBy(desc("RankNumber")).rowsBetween(Window.unboundedPreceding, Window.currentRow)

df = df.withColumn('previous_value',F.coalesce(F.lag(df['value'],1).over(w1),df['value']))

这是你的代码:

newdf = df.select(df.RankNumber,df.DeptNumber,df.Value,df.previous_value,when( df.Value<=df.previous_value, df.Value) \
                        .otherwise(F.min(df.previous_value).over(w2)).alias('newValue'))

>>> newdf.show()
+----------+----------+-----+--------------+--------+
|RankNumber|DeptNumber|Value|previous_value|newValue|
+----------+----------+-----+--------------+--------+
|         5|     Dept2|  100|           100|     100|
|         4|     Dept2|  102|           100|     100|
|         3|     Dept2|  105|           102|     100|
|         2|     Dept2|  198|           105|     100|
|         1|     Dept2|  194|           198|     194|
|         5|     Dept1|  200|           200|     200|
|         4|     Dept1|  202|           200|     200|
|         3|     Dept1|  205|           202|     200|
|         2|     Dept1|  198|           205|     198|
|         1|     Dept1|  194|           198|     194|
+----------+----------+-----+--------------+--------+

【讨论】:

感谢@Vikrant 的回复,您做对了。说第 7 条记录 "4 , 202, 6" 将更新为第 6 条记录 "4 , 200, 6"。由于第 5 条记录的值小于第 6 条记录的值,这有意义吗? 谢谢。所以你期待问题中提到的输出。我认为这是可以做到的。必须再有一个滞后。一旦进入系统,我会尝试一下。 :-) 我明白你的意思,可能是这样的......将解决问题df = df.withColumn('prepre_value',F.coalesce(F.lag(df['previous_value'],1).over(w),df['previous_value'])) 但我们如何知道要应用的滞后数,因为这需要动态完成并且不能采取查看数据集的电话。感谢您的帮助 当然..不用担心。慢慢来 我指的是记录“rankNumber = 1, DeptNumber = Dept2”,newValue 为 194。而 RankNumber = 2 的前一行的 NewValue 为 100。没关系,我已将此标记为已回答并完成。我会整理其余的。非常感谢您的努力和帮助!非常感激!祝你有美好的一天

以上是关于使用特定条件更新数据框的主要内容,如果未能解决你的问题,请参考以下文章

Spark Join 数据框并有条件地更新列

从前一行和特定列值有效地更新熊猫数据框中的 NaN

如何测试数据帧最后一行中的特定值(使用-1)

pandas基于条件判断更新dataframe中特定数据列数值内容的值(Conditionally updating values in specific pandas Dataframe )

pandas基于条件判断更新dataframe中特定数据列数值内容的值(Conditionally updating values in specific pandas Dataframe )

从列表字典更新我的数据框