在spark中更新dataframe列

Question

看看新的spark dataframe api，不清楚是否可以修改dataframe列。

如何更改DataFrame的行X列Y中的值？

from pyspark.sql import functions as F

update_func = (F.when(F.col('update_col') == replace_val, new_value)
                .otherwise(F.col('update_col')))
df = df.withColumn('new_column_name', update_func)

如果要对列执行某些操作并创建添加到DataFrame的新列：

import pyspark.sql.functions as F
import pyspark.sql.types as T

def my_func(col):
    do stuff to column here
    return transformed_value

# if we assume that my_func returns a string
my_udf = F.UserDefinedFunction(my_func, T.StringType())

df = df.withColumn('new_column_name', my_udf('update_col'))

如果希望新列的名称与旧列相同，可以添加附加步骤：

df = df.drop('update_col').withColumnRenamed('new_column_name', 'update_col')

Answer 1

从pyspark.sql.functions导入col，并根据字符串（字符串a、字符串b、字符串c）将第五列更新为整数（0,1,2）到一个新的DataFrame中。

from pyspark.sql.functions import col, when 

data_frame_temp = data_frame.withColumn("col_5",when(col("col_5") == "string a", 0).when(col("col_5") == "string b", 1).otherwise(2))