用spark中DataFrame对数据进行去重、缺失值处理、异常值处理

用spark中DataFrame对数据进行去重、缺失值处理、异常值处理用spark中DataFrame对数据进行清洗1.准备工作配置环境importosfrompysparkimportSparkContext,SparkConffrompyspark.sqlimportSparkSessionimportpyspark.sql.functionsasfnJAVA_HOME=/root/bigdata/jdkPYSPARK_PYTHON=/miniconda2/envs/py365/bin/pythonos.environJAVA_HOME=JAVA_HOMEos.environPYSPARK_PYTHON=PYSPARK_PYTHONos.environPYSPARK_DRIVER_PYTHON=PYSPARK_PYTHONSPARK_APP_NAME=dataframetestSPARK_URL=spark:/192.168.19.137:7077conf=SparkConf()创建sparkconfig对象config=(spark.app.name,SPARK_APP_NAME),设置启动的spark的app名称，没有提供，将随机产生一个名称(spark.executor.memory,6g),设置该app启动时占用的内存用量，默认g(spark.master,SPARK_URL),sparkmaste的地址(spark.executor.cores,4),设Ssparkexecutor使用的CPU核心数conf.setAll(config)利用config对象，创建sparksessionspark=SparkSession.builder.config(conf=conf).getOrCreate()2.数据去重1删除重复数据1.1删除完全一样的数据1.2删除某些字段值完全相同的记录1.3删除无意义字段的重复值!df=spark.createDataFrame(1,144.5,5.9,33,M),(2,167.2,5.4,45,M),(3,124.1,5.2,23,F),(4,144.5,5.9,33,M),(5,133.2,5.7,54,F),(3,124.1,5.2,23,F),(5,129.2,5.3,42,M),id,weight,height,age,gender)1首先删除完全一样的记录删除第3条和第6条中的一条df2=df.dropDuplicates().count()结果：22其次，关键字段值完全一模一样的记录(在这个例子中，是指除了之外的列一模一样)第1条和第四条删除某些字段值完全一样的重复记录subset参数定义这些字段df3=df2.dropDuplicates(subset=cforcindf.columnsifc!=id).count()纟吉果：53有意义的重复记录去重之后，再看某个无意义字段的值是否有重复(在这个例子中，是是否重复)df3.agg(fn.count(id).alias(id_dount),fn.countDistinct(id).alias(distinct_id_count).collect()结果：ROW(id_count=5,distinct_id_count=4)4对于id这种无意义的列重复，添加另外一列自增new_id的值并比连续且比较大df3.withColumn(new_id,fn.monotonically_increasing_id().show()3.缺失值处理2.处理缺失值2.1对缺失值进行删除操作(行，列)2.2对缺失值进行填充操作(列的均值)2.3对缺失值对应的行或列进行标记!df_miss=spark.createDataFrame(1,143.5,5.6,28,M,100000),(2,167.2,5.4,45,M,None),(3,None,5.2,None,None,None),(4,144.5,5.9,33,M,None),(5,133.2,5.7,54,F,None),(6,124.1,5.2,None,F,None),(7,129.2,5.3,42,M,76000),id,weight,height,age,gender,income)1计算每条记录的缺失值情况df_miss.rdd.map(lambdarow:(rowid,sum(c=Noneforcinrow).collect()结果：(1,0),(2,1),(3,4),(4,1),(5,1),(6,2),(7,0)2计算各列的缺失情况百分比df_miss.agg(*(1-fn.count(c)/fn.count(*).alias(c+_missing)forcindf_miss.columns).show()3按照缺失值删除行threshold是根据一行记录中，缺失字段的百分比的定义缺失三个就丢弃)df_miss_no_income.dropna(thresh=3).show()4删除缺失值过于严重的列其实是先建一个)F,不要缺失值的列df_miss_no_income=df_miss.select(cforcindf_miss.columnsifc!=income)5填充缺失值，可以用llna来填充缺失值，对于bool类型、或者分类类型，可以为缺失值单独设置一个类型missing对于数值类型，可以用均值或者中位数等填充filln可以接收两种类型的参数：一个数字、字符串，这时整个Datae中所有的缺失值都会被填充为相同的值。也可以接收一个字典列名：值这样先计算均值，并组织成一个字典means=df_miss_no_income.agg(*fn.means(c).alias(c+_mean)forcindf_miss_no_income.columnsifc!=gender).toPandas().to_dict(record)0meansgender=missingdf_miss_no_income.fillna(means).show()4.异常值处理3、异常值处理异常值：不属于正常的值包含：缺失值，超过正常范围内的较大值或较小值分位数去极值*中位数绝对偏差去极值正态分布去极值上述三种操作的核心都是：通过原始数据设定一个正常的范围，超过此范围的就是一个异常值df_outliers=spark.createDataFrame(1,143.5,5.3,28),(2,154.2,5.5,45),(3,342.3,5.1,99),(4,144.5,5.5,33),(5,133.2,5.4,54),(6,124.1,5.1,21),(7,129.2,5.3,42),id,weight,height,age)求出每个字段的边界用4分位计算cols=weight,height,agebounds=forcolincols:quantiles=df_outliers.approxQuantile(col,0.25,0.75,0.05)IQR=quantiles1-quantiles0boundscol=quantilesO-1.5*IQR,quantiles1+1.5*IQR结果：age:-11.0,93.0,height:4.499999999999999,6.1000000000000005,weight:91.69999999999999,191.7approxQuantil力法接收三个参数：参数，列名；参数2：想要计算的分位点，可以是一个点，也可以是一个列表)和1之间的小数)，第三个参数是能容忍的误差，如果是)，代表百分百精确计算。outliers=df_outliers.select(*id+(df_outlierscboundsc1).alias(c+_o)forcincols).show()|id|weight_o|height_o|age_o|+-+|1|false|false|false|2|false|false|false|3|true|false|true|4|false|false|false|5|false|false|false|6|false|false|false|7|false|false|false|+与原始数据关联df_outliers=df_outliers.join(outliers,on=id)取出有问题的值df_outliers.filter(weight_o).select(id,weight).show()df_outliers.filter(age_o).select(id,age).show()|id|weight|3|342.3|+1+|id|age|+1+|3|99|+1+