pyspark之数据处理学习【数据去重】(1)

1.重复数据

例如

spark &＃61; SparkSession.builder.appName("dataDeal").getOrCreate()
df &＃61; spark.createDataFrame([(1, 144.5, 5.9, 33, &＃39;M&＃39;),
(2, 167.2, 5.4, 45, &＃39;M&＃39;),
(3, 124.1, 5.2, 23, &＃39;F&＃39;),
(4, 144.5, 5.9, 33, &＃39;M&＃39;),
(5, 133.2, 5.7, 54, &＃39;F&＃39;),
(3, 124.1, 5.2, 23, &＃39;F&＃39;),
(5, 129.2, 5.3, 42, &＃39;M&＃39;),
], [&＃39;id&＃39;, &＃39;weight&＃39;, &＃39;height&＃39;, &＃39;age&＃39;, &＃39;gender&＃39;]) 上面的数据中存在如下问题&＃xff1a;

有两行id等于3并且完全相同

id为1和4的两行是一样的数据&＃xff0c;只是id不同&＃xff0c;可以假定为是同一个人的数据

有两行的id等于5&＃xff0c;这看上去是一个异常数据&＃xff0c;因为他们看上去不像是同一个人的数据

检查是否有重复数据采用.distinct()方法

print (&＃39;Count of rows:{0}&＃39;.format(df.count()))
print (&＃39;Count of distinct rows:{0}&＃39;.format(df.distinct().count()))
Count of rows:7
Count of distinct rows:6

可以看到返回的两个值不等&＃xff0c;一个为6&＃xff0c;一个为7.所以&＃xff0c;可以判断出我们的数据集中有完全相同的行&＃xff08;即重复的数据&＃xff09;

移除重复的数据采用.dropDuplicates()方法

#移除重复的数据
df &＃61; df.dropDuplicates()
#查看去重后的数据
df.show()
| id|weight|height|age|gender|
&＃43;---&＃43;------&＃43;------&＃43;---&＃43;------&＃43;
|  5| 133.2|   5.7| 54|     F|
|  5| 129.2|   5.3| 42|     M|
|  1| 144.5|   5.9| 33|     M|
|  4| 144.5|   5.9| 33|     M|
|  2| 167.2|   5.4| 45|     M|
|  3| 124.1|   5.2| 23|     F|
&＃43;---&＃43;------&＃43;------&＃43;---&＃43;------&＃43;

通过结果可以看出&＃xff0c;删除了一行id为3的记录

接着可以通过重复之前的工作检查与id无关的重复数据

#对除id以外的列进行对比
print ("Count of ids:{0}".format(df.count()))
print ("Count of distinct ids:{0}".format(df.select([c for c in df.columns if c !&＃61; &＃39;id&＃39;]).distinct().count())) Count of ids:6

Count of distinct ids:5

可以继续使用.dropDuplicates()删除重复数据&＃xff0c;但是需要使用subset参数来指定只处理除id以外的列。subset参数指明.dropDuplicates()方法只查找subset参数指定的列

#去掉除id以外其他属性相同的数据
df &＃61; df.dropDuplicates(subset&＃61;[c for c in df.columns if c !&＃61; &＃39;id&＃39;])
df.show()
&＃43;---&＃43;------&＃43;------&＃43;---&＃43;------&＃43;
| id|weight|height|age|gender|
&＃43;---&＃43;------&＃43;------&＃43;---&＃43;------&＃43;
|  5| 133.2|   5.7| 54|     F|
|  1| 144.5|   5.9| 33|     M|
|  2| 167.2|   5.4| 45|     M|
|  3| 124.1|   5.2| 23|     F|
|  5| 129.2|   5.3| 42|     M|
&＃43;---&＃43;------&＃43;------&＃43;---&＃43;------&＃43;

从结果可以看出&＃xff0c;现在的数据没有任何一行是重复的&＃xff08;既没有完全相同的记录也没有除id以外相同的记录&＃xff09;

检测是否有重复的id

#计算id的总数和id的唯一数
import pyspark.sql.functions as fn
df.agg(fn.count(&＃39;id&＃39;).alias(&＃39;count&＃39;),
fn.countDistinct(&＃39;id&＃39;).alias(&＃39;distinct&＃39;)
).show()
&＃43;-----&＃43;--------&＃43;
|count|distinct|
&＃43;-----&＃43;--------&＃43;
|    5|       4|
&＃43;-----&＃43;--------&＃43;

.count()方法和.countDistinct()方法分别计算DataFrame的行数和id的唯一数。.alias()方法可以对返回的列指定一个别名。

从结果中可以看出&＃xff0c;总共5条记录&＃xff0c;但只有4个唯一id。假设id相同的数据是偶然事件&＃xff0c;异常值&＃xff0c;则将每一行给定一个唯一的id

#重新给每行分配id
df.withColumn(&＃39;new_id&＃39;,fn.monotonically_increasing_id()).show()
&＃43;---&＃43;------&＃43;------&＃43;---&＃43;------&＃43;-------------&＃43;
| id|weight|height|age|gender|       new_id|
&＃43;---&＃43;------&＃43;------&＃43;---&＃43;------&＃43;-------------&＃43;
|  5| 133.2|   5.7| 54|     F|  25769803776|
|  1| 144.5|   5.9| 33|     M| 171798691840|
|  2| 167.2|   5.4| 45|     M| 592705486848|
|  3| 124.1|   5.2| 23|     F|1236950581248|
|  5| 129.2|   5.3| 42|     M|1365799600128|
&＃43;---&＃43;------&＃43;------&＃43;---&＃43;------&＃43;-------------&＃43;

.monotonicallymonotonically_increasing_id()方法给每条记录提供一个唯一且递增的id