spark를 이용해서 Missing Data다루기

In [1]:

import findspark

In [2]:

findspark.init('/home/hak/spark-2.4.5-bin-hadoop2.7/')

In [3]:

from pyspark.sql import SparkSession

In [4]:

spark = SparkSession.builder.appName('MissingData').getOrCreate()

In [5]:

df = spark.read.csv('ContainsNull.csv',header=True, inferSchema=True)

In [6]:

df.show()

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp1| John| null|
|emp2| null| null|
|emp3| null|345.0|
|emp4|Cindy|456.0|
+----+-----+-----+

In [44]:

df.printSchema()

root
 |-- Id: string (nullable = true)
 |-- Name: string (nullable = true)
 |-- Sales: double (nullable = true)

In [7]:

df.na.drop().show()  #not availiable  

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp4|Cindy|456.0|
+----+-----+-----+

In [10]:

#두개이상 값을 갖고있는 행을 가져오고 싶으면
df.na.drop(thresh=2).show()

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp1| John| null|
|emp3| null|345.0|
|emp4|Cindy|456.0|
+----+-----+-----+

In [12]:

#subset으로 특정 열부분만 삭제할 수 있다.
df.na.drop(subset=['Sales']).show()

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp3| null|345.0|
|emp4|Cindy|456.0|
+----+-----+-----+

In [18]:

df.na.drop(how='any').show()
#하나라도 Null값이라면

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp4|Cindy|456.0|
+----+-----+-----+

In [17]:

df.na.drop(how='all').show()
#전체가 Null값이면

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp1| John| null|
|emp2| null| null|
|emp3| null|345.0|
|emp4|Cindy|456.0|
+----+-----+-----+

In [20]:

#Null 부분에 값을 넣고 싶으면
#fill안에 값이 스트링이므로 스키마가 String인 Null값인 곳만 채워진다.
df.na.fill('NEW VALUE').show()

+----+---------+-----+
|  Id|     Name|Sales|
+----+---------+-----+
|emp1|     John| null|
|emp2|NEW VALUE| null|
|emp3|NEW VALUE|345.0|
|emp4|    Cindy|456.0|
+----+---------+-----+

In [21]:

df.printSchema()

root
 |-- Id: string (nullable = true)
 |-- Name: string (nullable = true)
 |-- Sales: double (nullable = true)

In [22]:

#숫자를 넣으면 스키마가 숫자인 부분만 적용된다.
df.na.fill(0).show()

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp1| John|  0.0|
|emp2| null|  0.0|
|emp3| null|345.0|
|emp4|Cindy|456.0|
+----+-----+-----+

In [23]:

#특정 칼럼에만 Null값을 바꾸고 싶으면
df.na.fill('No Name',subset=['Name']).show()

+----+-------+-----+
|  Id|   Name|Sales|
+----+-------+-----+
|emp1|   John| null|
|emp2|No Name| null|
|emp3|No Name|345.0|
|emp4|  Cindy|456.0|
+----+-------+-----+

In [24]:

from pyspark.sql.functions import mean

In [27]:

mean_value = df.select(mean(df['Sales'])).collect()

In [36]:

mean_value[0]  
# mean_value[0][0]    <- 400.5 값만 가져오고 싶을 때

Out[36]:

Row(avg(Sales)=400.5)

In [37]:

mean_sales = mean_value[0][0]

In [32]:

df.na.fill(mean_sales, subset=['Sales']).show()   #평균값을 구하고 null값에 평균값을 대입

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp1| John|400.5|
|emp2| null|400.5|
|emp3| null|345.0|
|emp4|Cindy|456.0|
+----+-----+-----+

In [33]:

#한꺼번에 작성하는 방법
df.na.fill(df.select(mean(df['Sales'])).collect()[0][0],subset=['Sales']).show()

+----+-----+-----+
|  Id| Name|Sales|
+----+-----+-----+
|emp1| John|400.5|
|emp2| null|400.5|
|emp3| null|345.0|
|emp4|Cindy|456.0|
+----+-----+-----+