[PySpark] 모델링 준비하기 - 결측치

토마스 드라마브, 데니 리의 "PySpark 배우기"를 보고 배워나가는 과정을 기록한 글입니다 ♪

---

데이터가 깨끗한 상태에 있다는 것을 스스로 증명하거나, 테스트하기 전 까ㅣ는 데이터를 모델링에 사용하거나 지나치게 신뢰하면 안된다. 

중복값, 결측치, 이상치, 존재하지 않는 주소, 잘못된 전화번호 / 지역코드, 잘못된 데이터 등을 변환해 데이터를 깨끗하게 만들어야 한다.

 

이번 포스팅에서는 

- 중복, 미관찰값, 또는 아웃라이어 다루고 인식

- 통계, 상관관계에 관해 계산

- matpotlib, Bokeh를 이용한 데이터 시각화 

 

중복, 미관찰값, 아웃라이어 확인하기

* 데이터를 완전히 검증하거나 검증 결과에 만족하기 전까지는 데이터를 신뢰하거나 사용해선 안된다 

 

중복 값

 

>>> 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']
... )

 

- ID가 같은 행 2개 (Id=3)

- 1행과 4행은 ID 빼고 다 동일 → 같은 사람 아닐까?

- ID가 5인 행 2개 → 다른 값이 모두 다르기 때문에 다른 사람 아닐까?

 

 

 

지금은 샘플 데이터의 크기가 아주 작다. 하지만 몇백줄의 큰 데이터를 확인할 때는 어떻게 해야할까?

▶ 먼저 중복 값이 존재하는지 확인해야 할 것이다. 

>>> print('Count of rows: {0}'.format(df.count()))
Count of rows: 7

>>> print('Count of distinct rows: {0}'.format(df.distinct().count()))
Count of distinct rows: 6

행의 개수와 유일값 개수를 확인하면 중복값이 존재하는지 확인 가능하다. 

위 두 결과값이 다르다면 중복값이 있다고 확신할 수 있다. 

 

 

 

dropDuplicates() 

- 중복값 제거

>>> df = df.dropDuplicates()
>>> df.show()
+---+------+------+---+------+
| id|weight|height|age|gender|
+---+------+------+---+------+
|  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|
+---+------+------+---+------+

ID가 3인 행이 두개 있었으나 하나를 삭제했다. 

 

 

이제 ID에 대해 다른 중복값이 있는지 확인하자

>>> print('Count of ids: {0}'.format(df.count()))
Count of ids: 6

>>> print('Count of distinct ids: {0}'.format(
...     df.select([c for c in df.columns if c != 'id']).distinct().count())
... )
Count of distinct ids: 5

다른 행에서 중복값이 있음을 알 수 있다. 

 

 

dropDuplicates() 함수를 사용한다면 id 컬럼을 제외한 나머지 컬럼을 subset으로 명시해야 한다.

>>> df = df.dropDuplicates(
...     subset=[c for c in df.columns if c != 'id'])

 

subset 파라미터로 명시된 컬럼을 이용해 중복된 행을 찾는다. 

id를 제외한 다른 컬럼 값이 동일할 경우의 행만 제거한다. 

id가 같다면 제거하지 않는다. 

 

>>> df.show()
+---+------+------+---+------+                                                  
| id|weight|height|age|gender|
+---+------+------+---+------+
|  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|
+---+------+------+---+------+

 

결과 값을 보면 이전에 id만 달랐던 id 1과 4 중 4가 삭제된 걸 확인할 수 있다. 

거의 다 된 것 같으니 중복값을 한번 더 확인해보자.

전체 고유 id 개수를 한번에 계산하기 위해 .agg() 함수를 사용한다.

 

>>> import pyspark.sql.functions as fn

>>> df.agg(
...     fn.count('id').alias('count'),
...     fn.countDistinct('id').alias('distinct')).show()
+-----+--------+                                                                
|count|distinct|
+-----+--------+
|    5|       4|
+-----+--------+

 

id의 행 개수를 count라는 컬럼명으로 보이고, id의 유일값 개수를 distinct라는 컬럼명으로 보여준다. 

 

위 결과는 전체 다섯 개의 행이 있으나 고유 id는 네개라고 말한다. 

그런데 중복 행은 모두 제거했으므로 monotonically_increasing_id() 함수로 각 행에 고유한 새 Id를 부여한다. 

 

 

 

monotonically_increasing_id()

 

- 각 행에 고유한 값을 부여하면서 그 값을 증가시킨다. 

>>> df.withColumn('new_id', fn.monotonically_increasing_id()).show()
+---+------+------+---+------+-------------+                                    
| id|weight|height|age|gender|       new_id|
+---+------+------+---+------+-------------+
|  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|
+---+------+------+---+------+-------------+

 

 

관찰되지 않은 데이터 

- 종종 아무 값도 없는 결측치 데이터를 볼 수 있다. 

- 시스템 장애, 휴먼에러, 데이터 스키마 변경과 같은 다양한 이유가 있을 수 있다. 

- 해결 방법? 

   1. 결측값 행 모두 제거 → 너무 많이 제거되지 않도록 주의, 너무 많이 제거하면 사용 불가할 수 있다

   2. None 값으로 채운다. 

      · 데이터가 참/거짓으로 구분되면 missing이라는 세번째 카테고리 생성 (0, 1, missing)

      · 데이터가 이미 카테고리를 갖고 있다면 missing 카테고리 추가 (0, 1, 2, 3... , missing)

      · 순서, 숫자 데이터를 갖고 있을 경우엔 평균, 중간값 또는 이외 정의된 다른 값으로 대체 가능 

 

예제 데이터 생성

>>> 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'])

행 분석

- ID가 3인 행은 height 외엔 모두 None 값이다. 

- ID가 6인 행은 오직 하나의 결측치(age)만 갖는다. 

 

컬럼분석

- income 컬럼은 매우 개인적인 정보이므로 거의 미관찰되었다. 

- wieght, gender 컬럼에선 미관찰 값이 존재하는 데이터가 하나뿐이다. (id=3)

- age 컬럼은 두 개의 미관찰값을 갖는다. 

 

 

 

각 행의 미관찰 값 개수를 확인하자

  - id에 대해 각 행마다 결측치가 몇 개인지 세주세요~

>>> df_miss.rdd.map(
...     lambda row: (row['id'], sum([c == None for c in row]))).collect()
[(1, 0), (2, 1), (3, 4), (4, 1), (5, 1), (6, 2), (7, 0)]

→ id가 3인 행은 4개의 결측치를 갖는다. 

 

 

 

- 각 행에서 미관찰 값들에 대해 어떻게 처리할지 결정하기 위해 어떤 값이 미관찰되었는지 확인하자. 

>>> df_miss.where('id == 3').show()
+---+------+------+----+------+------+
| id|weight|height| age|gender|income|
+---+------+------+----+------+------+
|  3|  null|   5.2|null|  null|  null|
+---+------+------+----+------+------+

 

 

 

이제 각 컬럼에서 미관찰 값의 비율을 확인하자

>>> df_miss.agg(*[
...     (1 - (fn.count(c) / fn.count('*'))).alias(c + '_missing') for c in df_miss.columns]).show()
+----------+------------------+--------------+------------------+------------------+------------------+
|id_missing|    weight_missing|height_missing|       age_missing|    gender_missing|    income_missing|
+----------+------------------+--------------+------------------+------------------+------------------+
|       0.0|0.1428571428571429|           0.0|0.2857142857142857|0.1428571428571429|0.7142857142857143|
+----------+------------------+--------------+------------------+------------------+------------------+

→ 14%의 미관찰 값이 wieght, gender 컬럼에 있다는 걸 알 수 있다. 

     income 컬럼은 72%나 된다. 

     그럼 이제 어떻게 처리해야할지 판단이 서기 시작할 것이다. 

 

 

 

1. 대부분의 값이 미관찰인 income 피쳐를 제거하자

>>> df_miss_no_income = df_miss.select(
...     [c for c in df_miss.columns if c != 'income'])

>>> df_miss_no_income.show()
+---+------+------+----+------+
| id|weight|height| age|gender|
+---+------+------+----+------+
|  1| 143.5|   5.6|  28|     M|
|  2| 167.2|   5.4|  45|     M|
|  3|  null|   5.2|null|  null|
|  4| 144.5|   5.9|  33|     M|
|  5| 133.2|   5.7|  54|     F|
|  6| 124.1|   5.2|null|     F|
|  7| 129.2|   5.3|  42|     M|
+---+------+------+----+------+

→ 이제 id가 3인 행을 제거하지 않아도 된다. 다른 행의 같은 컬럼 값들을 이용해 대체 가능하기 때문이다. 

    그럼에도 불구하고 id가 3인 행을 제거하고 싶다면 dropna() 함수를 사용하자.

 

 

dropna()

- thresh : 각 행에서 제거할 수 있는 최소의 미관찰 값 개수를 임계치로 설정 가능

    * 데이터셋이 수십, 수백개의 feature를 갖고 있거나 미관찰 값에 대한 임계치를 넘은 행을 제거하고 싶을 경우 유용!

 

>>> df_miss_no_income.dropna(thresh=3).show()
+---+------+------+----+------+
| id|weight|height| age|gender|
+---+------+------+----+------+
|  1| 143.5|   5.6|  28|     M|
|  2| 167.2|   5.4|  45|     M|
|  4| 144.5|   5.9|  33|     M|
|  5| 133.2|   5.7|  54|     F|
|  6| 124.1|   5.2|null|     F|
|  7| 129.2|   5.3|  42|     M|
+---+------+------+----+------+

 

 

위에서 id가 3인 행을 제거했다면 이제는 대체하는 과정을 살펴보자.

 

 

 

fillna()

- 미관찰 값을 추정해 채우려면 사용한다.

- integer, float, string 타입을 지원한다. (+ long) 

- 전체 데이터 셋에서 미관찰 값은 그 값으로 채워진다. 또한 딕셔너리 타입도 지원한다. (ex. {'<colName>' : <value_to_impute>})

   * 딕셔너리 타입에서 value_to_impute도 integer, float, string 타입만 지원한다.

- 평균, 중간값 또는 다른 계산된 값으로 대체하려면 우선 그 값을 계산하고 그 값을 가지는 딕셔너리를 만든 후 fillna()에 전달해야 한다.

 

>>> means = df_miss_no_income.agg(
...     *[fn.mean(c).alias(c) for c in df_miss_no_income.columns if c != 'gender']
... ).toPandas().to_dict('records')[0]

→ 카테고리 값에 대해서는 평균을 구할 수 없ㅇ므로 gender 컬럼을 제외한다. 

     agg() 함수의 결과를 취해 pandas 데이터프레임으로 변환 후 딕셔너리 형태로 변환까지 데이터타입이 두번 변한다.

 

 

>>> means
{'id': 4.0, 
 'weight': 140.28333333333333, 
 'height': 5.471428571428571, 
 'age': 40.4
 }

→ pandas의 to_dict() 함수에 대한 파라미터는 위와 같은 딕셔너리를 생성하라고 지시한다. 

 

>>> means['gender'] = 'missing'

→ gender는 수치계산이 불가하므로 missing이라는 카테고리를 gender 피쳐에 추가했다.

 

>>> means
{'id': 4.0, 
 'weight': 140.28333333333333, 
 'height': 5.471428571428571, 
 'age': 40.4, 
 'gender': 'missing'}

→ gender 컬럼에 missing 값 까지 넣어준 후의 means 값이다. 

※ age 컬럼의 값이 40.4라도 값 대체할 때에는 df_miss_no_income.age 컬럼 타입이 유지되어 정수 타입으로 들어간다. 

 

 

>>> df_miss_no_income.fillna(means).show()
+---+------------------+------+---+-------+
| id|            weight|height|age| gender|
+---+------------------+------+---+-------+
|  1|             143.5|   5.6| 28|      M|
|  2|             167.2|   5.4| 45|      M|
|  3|140.28333333333333|   5.2| 40|missing|
|  4|             144.5|   5.9| 33|      M|
|  5|             133.2|   5.7| 54|      F|
|  6|             124.1|   5.2| 40|      F|
|  7|             129.2|   5.3| 42|      M|
+---+------------------+------+---+-------+