如何在pyspark使用的一行中修改一列值

Question

我想在userid = 22650984时更新值。如何在pyspark平台中执行此操作？感谢您的帮助。

>>>xxDF.select('userid','registration_time').filter('userid="22650984"').show(truncate=False)
18/04/08 10:57:00 WARN TaskSetManager: Lost task 0.1 in stage 57.0 (TID 874, shopee-hadoop-slave89, executor 9): TaskKilled (killed intentionally)
18/04/08 10:57:00 WARN TaskSetManager: Lost task 11.1 in stage 57.0 (TID 875, shopee-hadoop-slave97, executor 16): TaskKilled (killed intentionally)
+--------+----------------------------+
|userid  |registration_time           |
+--------+----------------------------+
|22650984|270972-04-26 13:14:46.345152|
+--------+----------------------------+

Answer 1

如果要修改DataFrame的子集并保持其余部分不变，最好的选择是pyspark.sql.functions.when()使用filter或pyspark.sql.functions.where()删除所有不满足条件的行。

from pyspark.sql.functions import col, when

valueWhenTrue = None  # for example

df.withColumn(
    "existingColumnToUpdate",
    when(
        col("userid") == 22650984,
        valueWhenTrue
    ).otherwise(col("existingColumnToUpdate"))
)

何时将第一个参数评估为布尔条件。如果条件为True，它将返回第二个参数。您可以将多个when语句链接在一起，如本博文以及本博文所示。或者用于otherwise()指定条件为时的处理方式False。

在此示例中，我将更新现有的column "existingColumnToUpdate"。当userid等于指定值时，我将使用更新列valueWhenTrue。否则，我们将保持列中的值不变。

Answer 2

您可以使用以下方法withColumn来实现您想要做的事情：

new_df = xxDf.filter(xxDf.userid = "22650984").withColumn(xxDf.field_to_update, <update_expression>)

update_expression 将包含您的更新逻辑 - 可以是 UDF 或派生字段等。