dataframe添加一列_在PySpark数据框中添加新列的5种方法

每天都在生成太多数据。尽管有时我们可以使用Rapids或Parallelization等工具来管理大数据&＃xff0c;但如果您使用的是TB级数据&＃xff0c;Spark是一个很好的工具。尽管这篇文章解释了如何使用RDD和基本的Dataframe操作&＃xff0c;但是我在使用PySpark Dataframes时错过了很多东西。只有当我需要更多功能时&＃xff0c;我才阅读并提出多种解决方案来做一件事情。如何在Spark中创建新列&＃xff1f;现在&＃xff0c;这听起来微不足道&＃xff0c;但请相信我&＃xff0c;事实并非如此。您可能想要处理这么多数据&＃xff0c;所以我很确定您最终将在工作流中使用大多数这些列创建过程。有时使用Pandas功能&＃xff0c;有时使用基于RDD的分区&＃xff0c;有时使用成熟的python生态系统。这篇文章将是关于"在Pyspark Dataframe中创建新列的多种方法"。如果您安装了PySpark&＃xff0c;则可以跳过下面的"入门"部分。Spark入门我知道很多人不会在系统中安装Spark来尝试和学习。但是安装Spark本身就是一件令人头疼的事情。由于我们想了解它是如何工作的以及如何使用它&＃xff0c;因此建议您在此处与社区版一起在线使用Databricks上的Spark。不用担心&＃xff0c;它是免费的&＃xff0c;尽管资源较少&＃xff0c;但是对于我们来说&＃xff0c;出于学习目的&＃xff0c;它现在就适用。

一旦注册并登录&＃xff0c;将显示以下屏幕。

您可以在此处启动新笔记本。选择Python笔记本&＃xff0c;并为笔记本命名。启动新笔记本并尝试执行任何命令后&＃xff0c;笔记本将询问您是否要启动新群集。做吧下一步将检查sparkcontext是否存在。要检查sparkcontext是否存在&＃xff0c;您必须运行以下命令&＃xff1a;sc

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

数据在这里&＃xff0c;我将处理Movielens ml-100k.zip数据集。1000位用户观看1700部电影时获得100,000个评分。在此压缩文件夹中&＃xff0c;我们将专门使用的文件是评估文件。该文件名保留为" u.data"如果要上载此数据或任何数据&＃xff0c;可以单击左侧的"数据"选项卡&＃xff0c;然后使用提供的GUI添加数据。

然后&＃xff0c;我们可以使用以下命令加载数据&＃xff1a;

ratings &＃61; spark.read.load("/FileStore/tables/u.data",format&＃61;"csv", sep&＃61;"\t", inferSchema&＃61;"true", header&＃61;"false")ratings &＃61; ratings.toDF(*[&＃39;user_id&＃39;, &＃39;movie_id&＃39;, &＃39;rating&＃39;, &＃39;unix_timestamp&＃39;])外观如下&＃xff1a;

ratings.show()

好的&＃xff0c;现在我们准备开始我们感兴趣的部分。如何在PySpark Dataframe中创建一个新列&＃xff1f;

使用Spark本机函数在PySpark DataFrame中创建新列的最pysparkish方法是使用内置函数。这是创建新列的最高效的编程方式&＃xff0c;因此&＃xff0c;这是我想进行某些列操作时首先要去的地方。我们可以将.withcolumn与PySpark SQL函数一起使用来创建新列。本质上&＃xff0c;您可以找到已经使用Spark函数实现的String函数&＃xff0c;Date函数和Math函数。我们可以将spark函数导入为&＃xff1a;

import pyspark.sql.functions as F我们的第一个函数F.col函数使我们可以访问列。因此&＃xff0c;如果我们想将一栏乘以2&＃xff0c;可以将F.col用作&＃xff1a;

ratings_with_scale10 &＃61; ratings.withColumn("ScaledRating", 2*F.col("rating"))ratings_with_scale10.show()

我们还可以使用数学函数&＃xff0c;例如F.exp函数&＃xff1a;

ratings_with_exp &＃61; ratings.withColumn("expRating", 2*F.exp("rating"))ratings_with_exp.show()

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

Spark UDF有时我们想对一列或多列做复杂的事情。可以将其视为对PySpark数据框到单列或多列的映射操作。尽管Spark SQL函数确实解决了许多有关创建列的用例&＃xff0c;但只要我想使用更成熟的Python功能时&＃xff0c;我都会使用Spark UDF。要使用Spark UDF&＃xff0c;我们需要使用F.udf函数将常规的python函数转换为Spark UDF。我们还需要指定函数的返回类型。在此示例中&＃xff0c;返回类型为StringType()

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

使用RDD有时&＃xff0c;Spark UDF和SQL函数对于特定用例而言都是不够的。您可能想利用Spark RDD获得的更好的分区。或者&＃xff0c;您可能想在Spark RDD中使用组函数。您可以使用此方法&＃xff0c;主要是在需要访问python函数内部spark数据框中的所有列时。无论如何&＃xff0c;我发现使用RDD创建新列的这种方式对于有经验的RDD(这是Spark生态系统的基本组成部分)的人们非常有用。下面的过程利用该功能在Row和pythondict对象之间进行转换。我们将行对象转换为字典。按照我们的习惯使用字典&＃xff0c;然后将该字典再次转换回行。

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

Pandas UDFSpark版本2.3.1中引入了此功能。这使您可以在Spark中使用Pands功能。我通常在需要在Spark数据帧上运行groupby操作或需要创建滚动功能并想使用Pandas滚动功能/窗口功能的情况下使用它。我们使用它的方式是使用F.pandas_udf装饰器。我们在这里假设该函数的输入将是一个熊猫数据框。我们需要从该函数依次返回一个Pandas数据框。这里唯一的复杂性是我们必须为输出数据框提供一个架构。我们可以使用以下格式来实现。

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

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

# 0. Declare the schema for the output of our functionoutSchema &＃61; StructType([StructField(&＃39;replication_id&＃39;,IntegerType(),True),StructField(&＃39;RMSE&＃39;,DoubleType(),True)])# decorate our function with pandas_udf decorator&＃64;F.pandas_udf(outSchema, F.PandasUDFType.GROUPED_MAP)defrun_model(pdf): # 1. Get hyperparam values num_trees &＃61; pdf.num_trees.values[0] depth &＃61; pdf.depth.values[0] replication_id &＃61; pdf.replication_id.values[0] # 2. Train test split Xtrain,Xcv,ytrain,ycv &＃61; train_test_split..... # 3. Create model using the pandas dataframe clf &＃61; RandomForestRegressor(max_depth &＃61; depth, num_trees&＃61;num_trees,....) clf.fit(Xtrain,ytrain) # 4. Evaluate the model rmse &＃61; RMSE(clf.predict(Xcv,ycv) # 5. return results as pandas DF res &＃61;pd.DataFrame({&＃39;replication_id&＃39;:replication_id,&＃39;RMSE&＃39;:rmse}) return resresults &＃61; replicated_data.groupby("replication_id").apply(run_model)以上只是一个想法&＃xff0c;而不是一个有效的代码。尽管应该稍作修改。

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

ratings.registerTempTable(&＃39;ratings_table&＃39;)newDF &＃61; sqlContext.sql(&＃39;select *, 2*rating as newCol from ratings_table&＃39;)newDF.show()

希望我已经很好地介绍了列创建过程&＃xff0c;以帮助您解决Spark问题。

- END -

文源网络&＃xff0c;仅供学习之用&＃xff0c;如有侵权&＃xff0c;联系删除。往期精彩

◆  50款开源工具你都用过吗&＃xff1f;

◆  python&＃43;C、C&＃43;&＃43;混合编程的应用

◆  python网络爬虫的基本原理详解

◆  Python自动操控excel&＃xff0c;一小时解决你一天的工作

◆  如何用Python增强Excel&＃xff0c;减少处理复杂数据的痛苦&＃xff1f;