将函数应用于 Spark 中 csv 的单列

Question

我正在使用 Spark 读取一个 csv，并想将函数应用于 csv 上的列。我有一些有效的代码，但它非常 hacky。执行此操作的正确方法是什么？

我的代码

SparkContext().addPyFile("myfile.py")
spark = SparkSession\
    .builder\
    .appName("myApp")\
    .getOrCreate()
from myfile import myFunction

df = spark.read.csv(sys.argv[1], header=True,
    mode="DROPMALFORMED",)
a = df.rdd.map(lambda line: Row(id=line[0], user_id=line[1], message_id=line[2], message=myFunction(line[3]))).toDF()

我希望能够只调用列名上的函数，而不是将每一行映射到 line，然后再调用 line[index] 上的函数。

我使用的是 Spark 2.0.1 版

Answer 1

您可以简单地使用用户定义函数 (udf) 结合 withColumn :

from pyspark.sql.types import IntegerType
from pyspark.sql.functions import udf

udf_myFunction = udf(myFunction, IntegerType()) # if the function returns an int
df = df.withColumn("message", udf_myFunction("_3")) #"_3" being the column name of the column you want to consider

这将向数据框 df 添加一个新列，其中包含 myFunction(line[3]) 的结果。

将函数应用于 Spark 中 csv 的单列

Apply a function to a single column of a csv in Spark

apache-spark

pyspark

spark-dataframe