使用特定条件更新数据框
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。没关系，我已将此标记为已回答并完成。我会整理其余的。非常感谢您的努力和帮助！非常感激！祝你有美好的一天以上是关于使用特定条件更新数据框的主要内容，如果未能解决你的问题，请参考以下文章
pandas基于条件判断更新dataframe中特定数据列数值内容的值（Conditionally updating values in specific pandas Dataframe ）
pandas基于条件判断更新dataframe中特定数据列数值内容的值（Conditionally updating values in specific pandas Dataframe ）