[PySpark] 트랜스포메이션 - map(), filter(), flatMap(), distinct()

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

---

트랜스포메이션

트랜스포메이션은 데이터셋의 형태를 만듦 

→ 필터링, 조인, 데이터셋 내 값들에 대한 트랜스코딩을 포함

 

RDD에는 스키마가 없기 때문에 생산된 데이터셋에서의 스키마를 잘 알고있다고 가정함. 

파싱된 데이터셋에서 정보 위치는 이전 텍스트 파일 로드 포스팅의 extractInformation() 함수를 참고!

 

 

 

map() 트랜스포메이션 

 

- 사망 날짜를 숫자 값으로 변형하는 새로운 데이터셋 생성

>>> data_2014 = data_from_file_conv.map(lambda row: int(row[16]))
>>> data_2014.take(10)

예시로, 2014U7CN 라고 입력되어있는 사망 날짜를 아래와 같이 변형한다. 

 

[2014, 2014, 2014, 2014, 2014, 2014, 2014, 2014, 2014, -99]

 

 

- 17번째 행 엘리먼트 추가

>>> data_2014_2 = data_from_file_conv.map(lambda row: (row[16], int(row[16])))
>>> data_2014_2.take(10)

 

[('2014', 2014), ('2014', 2014), ('2014', 2014), ('2014', 2014), ('2014', 2014), ('2014', 2014), ('2014', 2014), ('2014', 2014), ('2014', 2014), ('-99', -99)]

 

 

 

.filter() 트랜스포메이션 

- 자주 쓰이는 트랜스포메이션

- 데이터셋으로부터 특정 조건에 맞는 엘리먼트 선택 가능

- ex) data_from_file_conv 데이터셋에서 2014년에 사고로 몇명이 죽었는지 카운트 

 

>>> data_filtered = data_from_file_conv.filter(
...     lambda row: row[16] == '2014' and row[21] == '0')

→ row[16]은 연도에 해당, row[21]은 생존 여부에 해당. (0 = 사망, 1 = 생존)

 

>>> data_filtered.count()
22

→ 총 22명이 2014년에 사망했다.

 

 

.flatMap(...) 트랜스포메이션

- map()과 비슷하게 동작

- 차이점은 리스트 반환이 아닌 평면화된 결과 반환

- 올바르지 않은 형태의 데이터를 제거할 때 사용

- 각 행을 리스트로 읽어서 추가

- 올바르지 않은 형태의 데이터는 빈 리스트를 전달해 제거함

 

>>> data_2014_flat = data_from_file_conv.flatMap(lambda row: (row[16], int(row[16]) + 1))

>>> data_2014_flat.take(10)
['2014', 2015, '2014', 2015, '2014', 2015, '2014', 2015, '2014', 2015]

결과값이 깔끔하게 반환되었다. 

그럼 map()의 결과물과 비교해보자. 

 

>>> data_2014_map = data_from_file_conv.map(lambda row: (row[16], int(row[16]) + 1))
>>> data_2014_map.take(10)
[('2014', 2015), ('2014', 2015), ('2014', 2015), ('2014', 2015), ('2014', 2015), ('2014', 2015), ('2014', 2015), ('2014', 2015), ('2014', 2015), ('-99', -98)]

위 값은 올바르지 않은 값 ('-99', -98)이 포함되어 있음을 볼 수 있다. 

 

 

 

 

.distinct() 트랜스포메이션

- 특정 컬럼에서의 중복 값 제거, 고유한 값을 리스트로 리턴 → 데이터셋 알아보거나 검증 시 유용함

- 많은 자원을 활용하는 함수이며 데이터를 섞는 연산을 포함하므로 정말 필요한 경우에만 사용하자

- ex) 성별 컬럼(row[5])이 남자, 여자만 포함하는지 알아보면, 제대로 파싱했는지를 검증하게 됨

>>> distinct_gender = data_from_file_conv.map(
...     lambda row: row[5]).distinct()

map()을 통해 성별 컬럼 추출 후 distinct() 함수로 고유한 값으로 어떤 값이 있는지 확인 

 

>>> distinct_gender.collect()
['-99', 'M', 'F']

collect() 함수로 고유값 출력