pyspark - 如何从 CSV 文件中清除数据
问题描述
样本名称.csv 数据:
Name, ,Age, ,Class,
Diwakar,, ,25,, ,12,
, , , , ,
Prabhat, ,27, ,15,
Zyan, ,30, ,17,
Jack, ,35, ,21,
读取 csv 文件:
names = spark.read.csv("name.csv", header="true", inferSchema="true")
names.show()
将此作为输出,我们正在丢失一些数据:
+-------+----+---+---+-----+----+
| Name| 1|Age| 3|Class| _c5|
+-------+----+---+---+-----+----+
|Diwakar|null| | 25| null| |
| | | | | |null|
|Prabhat| | 27| | 15|null|
| Zyan| | 30| | 17|null|
| Jack| | 35| | 21|null|
+-------+----+---+---+-----+----+
我想要一个如下所示的输出:
+-------+---+---+---+-----+----+
| Name| 1|Age| 3|Class| _c5|
+-------+---+---+---+-----+----+
|Diwakar| | 25| | 12|null|
| | | | | |null|
|Prabhat| | 27| | 15|null|
| Zyan| | 30| | 17|null|
| Jack| | 35| | 21|null|
+-------+---+---+---+-----+----+
解决方案
我们可以通过定义读取所有字段schema
,然后使用schema
while reading CSV 文件,然后使用When Otherwise
我们可以获得Age,Class
列的数据。
Example:
from pyspark.sql.functions import *
from pyspark.sql.types import *
#define schema with same number of columns in csv file
sch=StructType([
StructField("Name", StringType(), True),
StructField("1", StringType(), True),
StructField("Age", StringType(), True),
StructField("3", StringType(), True),
StructField("Class", StringType(), True),
StructField("_c5", StringType(), True),
StructField("_c6", StringType(), True)
])
#reading csv file with schema
df=spark.read.schema(sch).option("header",True).csv("name.csv")
df.withColumn('Age', when(length(trim(col('Age'))) == 0, col('3')).otherwise(col('Age'))).\
withColumn('1',lit("")).\
withColumn('3',lit("")).\
withColumn('Class',when((col('Class').isNull())|(lower(col('Class')) == 'null'), col('_c6')).when(length(trim(col('Class'))) == 0, lit("null")).otherwise(col('Class'))).\
withColumn('_c5',lit("null")).\
drop("_c6").\
show()
#+-------+---+---+---+-----+----+
#| Name| 1|Age| 3|Class| _c5|
#+-------+---+---+---+-----+----+
#|Diwakar| | 25| | 12|null|
#| | | | | null|null|
#|Prabhat| | 27| | 15|null|
#| Zyan| | 30| | 17|null|
#| Jack| | 35| | 21|null|
#+-------+---+---+---+-----+----+
推荐阅读
- java - 转换错误。parseInt 似乎没有转换我的文本值
- python - 我应该如何为 Django Channels 2.0 配置 Redis?
- php - 匹配 SQL Id 并获取并应用匹配值
- silverstripe - Silverstripe 不使用 HomePage.ss 作为 Chrome?
- javascript - 我的 AngularJS 应用程序中的“ng-repeat”有什么问题?
- php - 空或非空允许使用 php 插入
- watson-conversation - 问槽后如何在IBM沃森助手中捕获一个实体的多个值?
- java - 使用泛型:用泛型中的父值和子值实例化泛型类的对象
- java - 将可见性设置为 VISIBLE 时不显示进度条
- javascript - 如何编写一个接受参数并将其添加到数组的函数?