而且，它们都是有用的

> Image by nickgesell from Pixabay

每天都在生成太多数据。

尽管有时我们可以使用Rapids或Parallelization等工具来管理大数据，但如果您使用的是TB级数据，Spark是一个很好的工具。

在上一篇关于Spark的文章中，我解释了如何使用PySpark RDD和Dataframe。

尽管这篇文章解释了如何使用RDD和基本的Dataframe操作，但是我在使用PySpark Dataframes时错过了很多东西。

只有当我需要更多功能时，我才阅读并提出多种解决方案来做一件事情。

如何在Spark中创建新列？

现在，这听起来微不足道，但请相信我，事实并非如此。 您可能想要处理这么多数据，所以我很确定您最终将在工作流中使用大多数这些列创建过程。 有时使用Pandas功能，有时使用基于RDD的分区，有时使用成熟的python生态系统。

这篇文章将是关于"在Pyspark Dataframe中创建新列的多种方法"。

如果您安装了PySpark，则可以跳过下面的"入门"部分。

Spark入门

我知道很多人不会在系统中安装Spark来尝试和学习。 但是安装Spark本身就是一件令人头疼的事情。

由于我们想了解它是如何工作的以及如何使用它，因此建议您在此处与社区版一起在线使用Databricks上的Spark。 不用担心，它是免费的，尽管资源较少，但是对于我们来说，出于学习目的，它现在就适用。

一旦注册并登录，将显示以下屏幕。

您可以在此处启动新笔记本。

选择Python笔记本，并为笔记本命名。

启动新笔记本并尝试执行任何命令后，笔记本将询问您是否要启动新群集。 做吧

下一步将检查sparkcontext是否存在。 要检查sparkcontext是否存在，您必须运行以下命令：

sc

这意味着我们已经设置了可以运行Spark的笔记本。

数据

在这里，我将处理Movielens ml-100k.zip数据集。 1000位用户观看1700部电影时获得100,000个评分。 在此压缩文件夹中，我们将专门使用的文件是评估文件。 该文件名保留为" u.data"

如果要上载此数据或任何数据，可以单击左侧的"数据"选项卡，然后使用提供的GUI添加数据。

然后，我们可以使用以下命令加载数据：

ratings = spark.read.load("/FileStore/tables/u.data",format="csv", sep="", inferSchema="true", header="false")ratings = ratings.toDF(*['user_id', 'movie_id', 'rating', 'unix_timestamp'])

外观如下：

ratings.show()

好的，现在我们准备开始我们感兴趣的部分。 如何在PySpark Dataframe中创建一个新列？

1.使用Spark本机函数

> Photo by Andrew James on Unsplash

在PySpark DataFrame中创建新列的最pysparkish方法是使用内置函数。 这是创建新列的最高效的编程方式，因此，这是我想进行某些列操作时首先要去的地方。

我们可以将.withcolumn与PySpark SQL函数一起使用来创建新列。 本质上，您可以找到已经使用Spark函数实现的String函数，Date函数和Math函数。 我们可以将spark函数导入为：

import pyspark.sql.functions as F

我们的第一个函数F.col函数使我们可以访问列。 因此，如果我们想将一栏乘以2，可以将F.col用作：

ratings_with_scale10 = ratings.withColumn("ScaledRating", 2*F.col("rating"))ratings_with_scale10.show()

我们还可以使用数学函数，例如F.exp函数：

ratings_with_exp = ratings.withColumn("expRating", 2*F.exp("rating"))ratings_with_exp.show()

此模块中提供了许多其他功能，足以满足大多数简单的用例。 您可以在此处查看功能列表。

2. Spark UDF

> Photo by Divide By Zero on Unsplash

有时我们想对一列或多列做复杂的事情。 可以将其视为对PySpark数据框到单列或多列的映射操作。 尽管Spark SQL函数确实解决了许多有关创建列的用例，但只要我想使用更成熟的Python功能时，我都会使用Spark UDF。

要使用Spark UDF，我们需要使用F.udf函数将常规的python函数转换为Spark UDF。 我们还需要指定函数的返回类型。 在此示例中，返回类型为StringType()

import pyspark.sql.functions as Ffrom pyspark.sql.types import *  def somefunc(value): if value < 3:   return 'low' else:   return 'high'#convert to a UDF Function by passing in the function and return type of function udfsomefunc = F.udf(somefunc, StringType())ratings_with_high_low = ratings.withColumn("high_low", udfsomefunc("rating"))ratings_with_high_low.show()

3.使用RDD

> Photo by Ryan Quintal on Unsplash

有时，Spark UDF和SQL函数对于特定用例而言都是不够的。 您可能想利用Spark RDD获得的更好的分区。 或者，您可能想在Spark RDD中使用组函数。 您可以使用此方法，主要是在需要访问python函数内部spark数据框中的所有列时。

无论如何，我发现使用RDD创建新列的这种方式对于有经验的RDD(这是Spark生态系统的基本组成部分)的人们非常有用。

下面的过程利用该功能在Row和pythondict对象之间进行转换。 我们将行对象转换为字典。 按照我们的习惯使用字典，然后将该字典再次转换回行。

import mathfrom pyspark.sql import Rowdef rowwise_function(row): # convert row to dict:   row_dict = row.asDict() # Add a new key in the dictionary with the new column name and value.   row_dict['Newcol'] = math.exp(row_dict['rating']) # convert dict to row:   newrow = Row(**row_dict) # return new row   return newrow# convert ratings dataframe to RDDratings_rdd = ratings.rdd# apply our function to RDD  ratings_rdd_new = ratings_rdd.map(lambda row: rowwise_function(row))# Convert RDD Back to DataFrameratings_new_df = sqlContext.createDataFrame(ratings_rdd_new)ratings_new_df.show()

4.Pandas UDF

> Photo by Pascal Bernardon on Unsplash

Spark版本2.3.1中引入了此功能。 这使您可以在Spark中使用Pands功能。 我通常在需要在Spark数据帧上运行groupby操作或需要创建滚动功能并想使用Pandas滚动功能/窗口功能的情况下使用它。

我们使用它的方式是使用F.pandas_udf装饰器。 我们在这里假设该函数的输入将是一个熊猫数据框。 我们需要从该函数依次返回一个Pandas数据框。

这里唯一的复杂性是我们必须为输出数据框提供一个架构。 我们可以使用以下格式来实现。

# Declare the schema for the output of our functionoutSchema = StructType([StructField('user_id',IntegerType(),True),StructField('movie_id',IntegerType(),True),StructField('rating',IntegerType(),True),StructField('unix_timestamp',IntegerType(),True),StructField('normalized_rating',DoubleType(),True)])# decorate our function with pandas_udf decorator@F.pandas_udf(outSchema, F.PandasUDFType.GROUPED_MAP)def subtract_mean(pdf):    # pdf is a pandas.DataFrame    v = pdf.rating    v = v - v.mean()    pdf['normalized_rating'] =v    return pdfrating_groupwise_normalization = ratings.groupby("movie_id").apply(subtract_mean)rating_groupwise_normalization.show()

我们还可以利用它在每个火花节点上训练多个单独的模型。 为此，我们复制数据并为每个复制提供一个键和一些训练参数，例如max_depth等。然后，我们的函数将使用熊猫Dataframe，运行所需的模型，然后返回结果。 结构如下所示。

# 0. Declare the schema for the output of our functionoutSchema = StructType([StructField('replication_id',IntegerType(),True),StructField('RMSE',DoubleType(),True)])# decorate our function with pandas_udf decorator@F.pandas_udf(outSchema, F.PandasUDFType.GROUPED_MAP)def run_model(pdf):    # 1. Get hyperparam values    num_trees = pdf.num_trees.values[0]    depth = pdf.depth.values[0]    replication_id = pdf.replication_id.values[0]    # 2. Train test split    Xtrain,Xcv,ytrain,ycv = train_test_split.....    # 3. Create model using the pandas dataframe    clf = RandomForestRegressor(max_depth = depth, num_trees=num_trees,....)    clf.fit(Xtrain,ytrain)    # 4. Evaluate the model    rmse = RMSE(clf.predict(Xcv,ycv)    # 5. return results as pandas DF    res =pd.DataFrame({'replication_id':replication_id,'RMSE':rmse})    return res                results = replicated_data.groupby("replication_id").apply(run_model)

以上只是一个想法，而不是一个有效的代码。 尽管应该稍作修改。

5.使用SQL

对于喜欢SQL的人，甚至可以使用SQL创建列。 为此，我们需要注册一个临时SQL表，然后使用带有附加列的简单选择查询。 一个人也可以用它来进行联接。

ratings.registerTempTable('ratings_table')newDF = sqlContext.sql('select *, 2*rating as newCol from ratings_table')newDF.show()

结论

> Photo by Kelly Sikkema on Unsplash

到此为止(双关语意)

希望我已经很好地介绍了列创建过程，以帮助您解决Spark问题。 如果您需要了解更多Spark基础知识，请查看：

不只是介绍

您可以在GitHub存储库或databricks上已发布的笔记本中找到此文章的所有代码。

另外，如果您想了解有关Spark和Spark DataFrames的更多信息，我想讲一门关于Big Data Essentials的优秀课程，这是Yandex提供的Big Data Specialization的一部分。

感谢您的阅读。 我将来也会写更多对初学者友好的文章。 在Medium中关注我或订阅我的博客以了解有关它们的信息。 与往常一样，我欢迎您提供反馈和建设性的批评，可以在Twitter @mlwhiz上与他们联系。

另外，这是一个小的免责声明-由于共享知识从来都不是一个坏主意，因此本文中可能会有一些与相关资源相关的会员链接。

(本文翻译自Rahul Agarwal的文章《5 Ways to add a new column in a PySpark Dataframe》，参考：https://towardsdatascience.com/5-ways-to-add-a-new-column-in-a-pyspark-dataframe-4e75c2fd8c08)