PySpark - 根据条件填充特定行

Posted 2023-04-15

【中文标题】PySpark - 根据条件填充特定行

【英文标题】：PySpark - Fillna specific rows based on condition

【发布时间】：2019-08-09 18:23:25

【问题描述】：

我想替换数据框中的空值，但仅限于符合特定条件的行。

我有这个数据框：

A|B   |C   |D   |
1|null|null|null|
2|null|null|null|
2|null|null|null|
2|null|null|null|
5|null|null|null|

我想这样做：

A|B   |C   |D   |
1|null|null|null|
2|x   |x   |x   |
2|x   |x   |x   |
2|x   |x   |x   |
5|null|null|null|

我的情况

因此，A 列中编号为 2 的所有行都应该被替换。

A、B、C、D 列是动态的，它们的数字和名称会发生​​变化。

我还希望能够选择所有行，而不仅仅是替换的行。

我尝试了什么

我尝试使用 df.where 和 fillna，但它没有保留所有行。

我也想用withColumn做，但我只知道A列，其他所有的都会在每次执行时改变。

适应的解决方案：

 df.select("A",
             *[
                 when(col("A") == '2', 
                    coalesce(col(c),
                    lit('0').cast(df.schema[c].dataType))
                 ).otherwise(col(c)).alias(c) 
                 for c in cols_to_replace
               ])

【参考方案1】：

将pyspark.sql.functions.when 与pyspark.sql.functions.coalesce 一起使用：

from pyspark.sql.functions import coalesce, col, lit, when

cols_to_replace = df.columns[1:]
df.select(
    "A",
    *[
        when(col("A")==2, coalesce(col(c), lit("x"))).otherwise(col(c)).alias(c) 
        for c in cols_to_replace
     ]
).show()
#+---+----+----+----+
#|  A|   B|   C|   D|
#+---+----+----+----+
#|  1|null|null|null|
#|  2|   x|   x|   x|
#|  2|   x|   x|   x|
#|  2|   x|   x|   x|
#|  5|null|null|null|
#+---+----+----+----+

在列表推导中，您检查A 的值是否为2。如果是，则合并列的值和文字x。这会将nulls 替换为x。否则，保持相同的列值。

【讨论】：

难以置信你的回复速度！我只是补充说我遇到了数据类型的问题。因此，为了解决这个问题，我将“x”替换为 0，并使用数据框模式从合并内部转换为任何类型。

df = (df.select("A",*[ when(col("A") == '2', coalesce(col(c), lit('0').cast (df.schema[c].dataType))).otherwise(col(c)).alias(c) for c in cols_to_replace ]))