pyspark基础学习——数据处理

admin管理员组

文章数量:1794759

pyspark基础学习——数据处理

目录

• 前言
• 一、准备工作和数据的导入选择
• • 1.1 导入数据
  • 1.2 选择数据子集：
  • 1.3 列名重命名
• 二、数据清洗
• • 2.1 检测空值数量
  • 2.2 删除存在空值的行
  • 2.3 forward，backward填充
• 三、 数据处理
• • 3.1 数据筛选
  • 3.2 数据统计
  • 3.3 数据类型转换
  • 3.4 采用SQL语法进行处理
• 四、数据导出
• 总结

---

前言

上一篇文章中讲了如何在windows下安装和检测: pyspark，同时简单介绍了运行的环境。本文想就我的一些学习经验，分享一下使用pyspark来处理csv文件上的一些常用的pyspark语法。

一、准备工作和数据的导入选择

运行python代码，第一件事当然是导入对应的包，同时我们要为spark先创建好相应的环境，并且，spark中支持SQL，而且在SQL中有众多的函数，因此我们可以创建SparkSession对象，为了后续SQL函数的调用，我们要导入functions包，以及数据类型转换的时候，我们要导入types的包。

import pyspark from pyspark.sql import SparkSession from pyspark.sql.types import * from pyspark.sql.functions import * from pyspark.sql.types import TimestampType spark = SparkSession.builder.getOrCreate() sc = spark.sparkContext 1.1 导入数据

将csv文件导入为Dataframe样式： header表示是否需要导入表头；inferSchema表示是否需要推导出数据的类型（false默认为string）；delimiter表示指定分隔符进行读取。file对应文件的位置。

df1 = spark.read.options(header='True', inferSchema='True', delimiter=',').csv(file) 1.2 选择数据子集：

drop中填入不需要的列的列名。

df2 = df1.drop('列名') 1.3 列名重命名 df3=df2.withColumnRenamed("original name", "modified name")

如果有多个列的列名要进行修改，可以直接在后面再加上withColumnRenamed()进行修改

二、数据清洗

因为数据本身的问题，在处理的过程中需要我们对一些空值、异常值等进行处理。但是此次作业获取到的数据中主要是对空值的处理，因此对于异常值的处理不进行讨论

2.1 检测空值数量 df3.toPandas().isnull().sum() 2.2 删除存在空值的行

对于一些关键列的数据丢失、或是该行的缺失值占比较高的情况下，我们很难将人工将其弥补，因此直接对该行进行删除。

df_clear=df3.dropna(subset='列名') 2.3 forward，backward填充

forward： 前面一个值填充后面 backward：后面一个值填充前面

代码示例：

df = spark.createDataFrame([ (1, 'd1',None), (1, 'd2',10), (1, 'd3',None), (1, 'd4',30), (1, 'd5',None), (1, 'd6',None), ],('id', 'day','temperature')) df.show()

运行结果如下：

iddaytemperature

1	d1	null
1	d2	10
1	d3	null
1	d4	30
1	d5	null
1	d6	null

from pyspark.sql.window import Window forward = Window.partitionBy('id').orderBy('day').rowsBetween( Window.unboundedPreceding, Window.currentRow) backward = Window.partitionBy('id').orderBy('day').rowsBetween( Window.currentRow, Window.unboundedFollowing) df.withColumn('forward_fill', last('temperature', ignorenulls=True).over(forward))\\ .withColumn('backward_fill', first('temperature', ignorenulls=True).over(backward))\\ .show()

填充后的结果如下表所示：

iddaytemperatureforward_fillbackward_fill

1	d1	null	null	10
1	d2	10	10	10
1	d3	null	10	30
1	d4	30	30	30
1	d5	null	30	null
1	d6	null	30	null

Window.unboundedPreceding:分区的开始位置 Window.currentRow:分区计算到现在的位置 Window.unboundedFollowing:分区的最后位置。 负数：表示若前面有元素，范围向前延申几个元素 0：表示当前位置，等价于Window.currentRow 正数：表示若后面有元素，范围向后延申几个元素

三、 数据处理 3.1 数据筛选 data1= df_clear.filter(df_clear['column'] == 'attribute') # 条件过滤 data2 = df_clear.select('column') # 选择某一列的数据 3.2 数据统计 # 输出树状结构（输出列名、数据类型和是否能为空值） df_clear.printSchema() # 将该列数据进行汇总统计 df_clear.select('column').describe().show() # 求平均，按照id的方式进行统计 ave_column = df_clear.groupBy('id').agg({'column': 'mean'})

agg（{“列名”，“函数名”}）为聚合函数，其中有：

函数名作用

avg	求均值
count	计数
max	求最大值
mean	求均值
min	求最小值
sum	求和

3.3 数据类型转换 from pyspark.sql.functions import * # 转换为Int类型 df_clear.withColumn("column",df.age.cast('int')) # 转换为String类型 df_clear.withColumn("column",df.age.cast('string')) # 转换为Data类型 df_clear= df_clear.withColumn('column', to_date(df_clear['column'])) # 转换为TimestampType类型 dfTime=df_clear.withColumn('column',F.col('column').cast(TimestampType())) 3.4 采用SQL语法进行处理 df_sql_cf=df_clear.createOrReplaceTempView("carflow") spark.sql("select * from carflow\\ where sum_Total_CF=\\ (select max(sum_Total_CF) from carflow)").show() 四、数据导出 # ascending表示是否为升序，默认为True df_clear_asc= df_clear.orderBy("column"，ascending=False) # 将对应的数据类型转化为list，再导出为csv文件 df_asc= df_clear_asc.select(F.collect_list('column')).first()[0] df_asc.select("col1","col2","col3").toPandas().to_csv("total.csv")

---

总结

由于此次学习仅用于完成课堂大作业，因此有不足之处还望各位大佬在评论区制指正，若是能够为你们提供一点小小的帮助，希望各位大佬们能动动手指，给小弟一个赞！感谢各位大佬们！ 该作业的处理的源代码和相关数据已经传至github

本文标签： 数据处理基础pyspark