案例背景

最近上班需要处理的都是百万,千万级的数据,pandas的性能已经不够看了(虽然它在处理数据上是真的很好用),公司都是用的polar和pyspark,我最近也学习了一些,然后写篇文章对比一下他们的常见用法。虽然他们都有数据框dataframe这个数据结构,但是具体用法还是有很多差异的。

数据选取

都是做数据分析的,那么就用最简单的机器学习数据集波士顿房价数据集吧,演示以下常见的数据处理的用法。,然后画个图,简单机器学习一下。

pyspark自己电脑装了也用不了,要分布式的虚拟机。。我这里spark都是用公司环境跑的。和pd和pl本地跑的不一样。

代码实现

我这里就用data1表示pandas的数据结构,data2表示polar数据结构。data3表示pyspark 的数据结构。3个数据结构的相同功能都用各种对应的语法展示。(data3用的不是波士顿房价的数据,是随便从公司sql里面找的一个数据....)

导入包:

import numpy as np

import pandas as pd

import polars as pl

import matplotlib.pyplot as plt

import seaborn as sns

plt.rcParams ['font.sans-serif'] ='SimHei' #显示中文

plt.rcParams ['axes.unicode_minus']=False #显示负号

数据读取

首先是pandas的

data1=pd.read_excel('Boston.xlsx')

data1.head(2)

这个其实polar差不多:

data2=pl.read_excel('Boston.xlsx') #要装一个什么csv2xlsx的包

data2.head(2)

pyspark也差不多,但是我读不了本地数据....一直路径不对,可能公司虚拟机里面的的pyspark 的环境不一样。

#df = spark.read.csv("path/to/csv/file.csv", header=True, inferSchema=True)

一般来说你都用pyspark了,肯定不会从本地csv读取数据,一般都是直接从sql里面掏:

pyspark里面的.head()是只展示一行数据。

一般他们都是这样show()

敏感信息打了个码。

写入文件

#data1.to_csv(filepath)

#data2.write_csv(filepath)

#data3.write.csv(filepath,header=True)

这里就没运行了,但是是可用的,语法差不多。

查看数据基本信息

pandas很方便的info就行

data1.info()

 

polar里面具有没有这个方法,见鬼,我只找到一个差不多的:  

data2.schema

这个只有数据类别,没得非空值什么的信息,差评。

pyspark是这样的:

data3.printSchema()

其他的类型,变量名,数据形状,都是差不多:

print(data1.dtypes) ;print(data2.dtypes) #类型

print(data1.shape) ;print(data2.shape) #形状

print(data1.columns);print(data2.columns) #列名称

pyspark是这样的:

print(data3.dtypes)

print((data3.count(), len(data3.columns)))

print(data3.columns)

描述性统计

这是也是一样的,都是data.describe()就行

data1.describe()

 

data2.describe()

pyspark是这样的

data3.describe().show()

查看缺失值和填充

pandas 的我很熟悉

print(data1.isnull().sum())

data1.fillna('2')

polars有些变化

print(data2.null_count())

data2.fill_null('2')

ps:polars填充了然后想覆盖掉原来的列还很麻烦,大概这样:

pdf_global_dev1.with_columns(pdf_global_dev1['col_name'].fill_null(0).alias('col_name'))

pyspark贼复杂

from pyspark.sql.functions import col, sum,expr,split,substring,when

data3.agg(*[sum(col(c).isNull().cast("int")).alias(c) for c in data3.columns]).show()

# 显示每列的缺失值数量

填充倒是一样的

data3.fillna(2)

数据切片

我觉得pandas 的数据切片真的很厉害,很逻辑完善。

data1.loc[10:15,['CRIM','ZN']]

polars还没怎么用熟,不好评价

data2.slice(10, 5).select(['CRIM', 'ZN']) ##10是开始,5是行数

这里就得提一下了,polars没有索引这个东西,所以他的切片是用开始的位置和长度来穿入参数的。太离谱了,pandas的多层索引简直是诺贝尔奖发现好不好,polars居然没得索引这个东西,虽然简化一些东西,但是很多功能都丢失了。

pyspark就更加离谱了,还需要加个辅助列表示行数才能切片。。。

from pyspark.sql.functions import monotonically_increasing_id

df_with_row_index = data3.withColumn("row_index", monotonically_increasing_id()) #新增辅助列

# 切片操作

df_with_row_index.filter((col("row_index") >= 10) & (col("row_index") <= 15)).drop("row_index").select("age", "gender").show()

可能pyspark这种对大数据处理的只会动列,不会动行,也就没得行索引这个东西,连切片都没对应的方法。

数据筛选

语法差不多,各有优缺点吧

pandas

data1[data1['MEDV'] >49.9 ]

 polars

data2.filter(data2['MEDV'] >49.9)

pyspark

data3.filter(data3.age>90).show()

选两列数据

data1[['NOX', 'RAD']]

data2.select(['NOX', 'RAD'])

data3.select("age", "gender")

数据合并

pandas 的方法太多了:

data1.merge(pd.Series(np.random.randint(0, 10, size=len(data1)),name='new'),left_index=True, right_index=True)

## 还可以pd.concat([]),还可以直接data1['new']=

polars这个名称是真离谱啊,合并了之后名称不知道怎么给,只能重命名

##### 也是新增一列的用法

data2.with_columns(pl.Series('new1',np.random.randint(0, 10, size=len(data1)))).rename({'new1':'new'})

 

 pyspark就是这样子加一列。

data3_with_new = data3.withColumn("new", expr("rand() * 10")) # 使用rand()函数生成随机数列

分组聚合

分组聚合麻烦起来很麻烦,可以写得超级复杂,咱们就简单求和试试

pandas:

data1.groupby('RAD').sum()

 polars差不多

data2.groupby('RAD').sum()

pyspark的

data3.groupBy('family_role_code').count().show()

删除缺失值

数据是没缺失值的,就随便演示一下了

#data1.dropna()

data2.drop_nulls()

spark和pandas一样

data3.dropna()

数据排序

pandas长一些

data1.sort_values(['MEDV','B'],ascending=False)

polars短一点,且参数名称可能有点不一样

data2.sort(by=['MEDV', 'B'],descending=True)

data3.orderBy([data3.age,data3.nik_num],ascending=False).show(5)

列名称重命名

panadas除了rename,其实还有很多方法的。

data1.rename(columns={'MEDV':'y'})

polars也有rename,但是不能传入columns,而且他也有很多别的方法

#data2.select(pl.col('MEDV').alias('10Medv')) #挑选一行数据出来重命名

data2.select([ pl.col('NOX').alias('no'),

pl.col('RAD').alias('yes')]) #挑选2行数据出来重命名

data2.rename({'MEDV':'y'}) #没索引这个概念,所以就不用传入形参

spark就是这样的:

data3.withColumnRenamed('kk','户口号')

应用函数

简单的函数应该是差不多的,复杂的可以有区别,但是后面遇到再说

data1['DIS'].apply(lambda x:str(x).split('.')[0])

data2['DIS'].apply(lambda x:str(x).split('.')[0])

 

添加一个是否包含的虚拟变量:

pdf_global_dev1= pdf_global_dev1.with_columns(

pl.col('rulelist').map_elements(lambda x : '1' if rule in x else '0', return_dtype=pl.Utf8).alias(f'{rule}').fill_null(0))

数值型就是这样的:

for rule in set_kind:

# 对每一条规则,创建一个新列,如果规则在'rulelist'列中,则该列值为1,否则为0

df_cust1 = df_cust1.with_column(

pl.col('rulelist').apply(lambda x : 1 if rule in x else 0, return_dtype=pl.Int64).alias(f'{rule}')

)

我顺便验证一下pyspark切换数据类型的语法,多写了几个方法函数:

这个是年龄变成字符串切片整数

data3.withColumn("age_int", split(col("age").cast("string"), "\\.").getItem(0)).show(10) # 取age 的整数

这是将年龄和nik_num变成整型数

data3=data3.withColumn("age", col("age").cast("integer"))#.show()

data3=data3.withColumn("nik_num", col("nik_num").cast("integer"))

data3.show()

这是将户口本变成数值型数据然后+4

data3.withColumn("kk_new", col("kk").cast("double") + 4).show(8) # 转为数值然后+1

这是 将身份证里面的2-4位切片出来(spark居然是从1开始,不是0开始,我人晕了)

data3.withColumn("nik_shi", substring(col("nik").cast("string"), 3, 2)).show() #

连接数据

pandas里面主要是merge函数

polars更像sql里面用join

#data1.merge(data11, on='key')

#data2.join(data22, on='key')

#data3.join(data3_with_new, data3["kk"] == data3_with_new["kk"])

#pyspark一般用法: #df1.join(df2, df1['col'] == df2['col'])

删除某一列数据

两个一模一样

data1.drop(columns=['MEDV'])

data2.drop(columns=['MEDV'])

pyspark可以不用传入形参,我先增加一列,然后删除一列。

data3.withColumn('newColumn',F.lit('new')).drop('newColumn')

 相互转化

polars数据结构也可以变成pandas和numpy的

import pyarrow as pa

data2.to_pandas() #需要安装pyarrow

#data2.to_numpy() #也可以直接运行

data3.toPandas()

pyspark变不成numpy,但变成了pandas再变numpy也是一样的。 

链式法则

弄个复杂一点的代码段吧,来对比他们的区别

pandas的

(data1.assign(sum_B=data1['B'].sum()) # 求和的

.assign(sorted_MEDV=data1['MEDV'].sort_values()) # 排序的

.assign(first_name=data1['RAD'].iloc[0]) # 第一个

.assign(Medv=data1['MEDV'].mean() * 10) # 10倍

)[['sum_B','sorted_MEDV','first_name','Medv']].sort_values('sorted_MEDV')

polars 的:  

data2.select(

pl.sum('B'), #求和的

pl.col('MEDV').sort(), #排序的

pl.col('RAD').first().alias('first name'), #第一个

(pl.mean('MEDV')*10).alias('10Medv'), #10倍

)

spark就不写了,报错弄了半天也不知道为什么。。

数量统计

panadas我最常用的就是value_counts()的方法了吧,很方便统计一列数据每个出现的次数。

data1['RAD'].value_counts()

polars好像也一样

pyspark的对应:  

#grouped_df['kk_ucnt'].value_counts()。 pandas方法

(grouped_df.groupBy('kk_ucnt')

.count()

.orderBy(F.col('count').desc()))

# grouped_df.groupby('nik_ucnt')['kk_ucnt'].value_counts().unstack()

result_df = (grouped_df.groupBy('nik_ucnt')

.pivot('kk_ucnt')

.count() # 计算每个组合的次数

.na.fill(0) # 将空值填充为0,相当于unstack后的fillna(0)

.orderBy('nik_ucnt'))

画图

pandas 的对象可以直接.plot画图的,也可以plt画图,我发现polars对象也能直接plt画图,还不错。

plt.figure(figsize=(3,2))

data1['MEDV'].plot.box() # 只有padnas对象可以这样这个方法

#plt.boxplot(data1['MEDV']) #正常的plt画图

plt.boxplot(data2['MEDV']) #pl数据也能用plt画图

plt.show()

spark没得直接画图的方法,也是变成了pandas才能画图

plt.figure(figsize=(3,2))

plt.boxplot(data3.select(['age']).toPandas()) #sparkde 数据只能转为pd才能画图

plt.show()

机器学习 

我们直接把polars数据结构扔到sklearn库里面去

from sklearn.model_selection import train_test_split

from sklearn.linear_model import LinearRegression

from sklearn.ensemble import RandomForestRegressor

X = data2.drop(columns=['MEDV'])

y = data2['MEDV']

# 划分训练集和测试集

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

线性回归:

# 线性回归

model = LinearRegression()

model.fit(X_train, y_train)

model.score(X_test, y_test)

随机森林回归

# 随机森林

model = RandomForestRegressor()

model.fit(X_train, y_train)

model.score(X_test, y_test)

 

居然是和pandas一模一样的数据结果,还不错。

pyspark那差别就大了:

我们先转化数据都变成数值型:

data3=data3.withColumn("kk", col("kk").cast("double")).withColumn("nik", col("nik").cast("double"))

data3=data3.withColumn("family_role_code", col("family_role_code").cast("integer"))

data4=data3.withColumn("gender", when(col("gender") == "F", 1).otherwise(0))

data4.groupBy("gender").count().show()

data4.agg(*[sum(col(c).isNull().cast("int")).alias(c) for c in data4.columns]).show()

 确定没缺失值后,进行机器学习:也是随机森林

from pyspark.ml.feature import StringIndexer, VectorAssembler

from pyspark.ml.classification import RandomForestClassifier

from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# 创建SparkSession对象

spark = SparkSession.builder.getOrCreate()

label_indexer = StringIndexer(inputCol="gender", outputCol="label")

data4=data4.na.drop()

# 特征向量转换器

feature_columns = data4.columns

feature_columns.remove('gender')

feature_assembler = VectorAssembler(inputCols=feature_columns, outputCol="features")

# 随机森林分类器

rf = RandomForestClassifier(featuresCol="features", labelCol="label")

# 构建Pipeline

from pyspark.ml import Pipeline

pipeline = Pipeline(stages=[label_indexer, feature_assembler, rf])

# 拆训练集和测试集

train_data, test_data = data4.randomSplit([0.7, 0.3], seed=123)

# 训练模型

model = pipeline.fit(train_data)

predictions = model.transform(test_data)

evaluator = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="accuracy")

accuracy = evaluator.evaluate(predictions)

print("准确率:", accuracy)

哇贼复杂,要变形编码数据后构建管道,感觉不好用。。不是数据量大到上千w我是不会想用这个来机器学习的。。。sklearn才是真神。

总结

感觉上来说,pandas和polars最大差异在于polars没得索引这个东西,并且很多  新增列,条件筛选,切片等等是存在一定的差异的,相似地方也有一些。最让我惊喜的是plt和sklearn可以完美兼容polars的数据结构,那就真的很不错了。

pyspark有的位置和pandas很像,但是有的位置又和他们差太远了,要用好这个东西还是得多练多写。

文章链接

评论可见,请评论后查看内容,谢谢!!!评论后请刷新页面。