[PySpark] 텍스트 파일 로드

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

---

1. 텍스트 파일 로드 

VS14MORT라는 파일을 다운로드한다. 

다운로드 주소는 http://tomdrabas.com/data/VS14MORT.txt.gz 를 들어가면 바로 다운로드가 진행된다. 

다운로드 후 압축을 풀고~ 

SparkContext의 textFile 함수로 텍스트 파일을 불러온다. 

 

data_from_file = sc.textFile('/Users/hayeon/Downloads/VS14MORT.txt', 4)

경로 뒤의 숫자 4는 데이터셋이 나눠진 파티션의 개수를 의미한다. 

 

 

가장 첫번째 행을 살펴보자

>>> data_from_file.take(1)

 

 

 

음 아주 보기 복잡하게 되어있군요~ 

 

 

2. 리스트로 변환

읽기 편하게 각 값이 리스트로 표현되게 만들자. 

 

a. 파싱

 

행을 읽을 수 있는 형태로 파싱하는 코드

각 행 하나씩 파싱하고 리턴하는 코드이다. 

 

>>> def extractInformation(row):
...     import re
...     import numpy as np
...     selected_indices = [
...          2,4,5,6,7,9,10,11,12,13,14,15,16,17,18,
...          19,21,22,23,24,25,27,28,29,30,32,33,34,
...          36,37,38,39,40,41,42,43,44,45,46,47,48,
...          49,50,51,52,53,54,55,56,58,60,61,62,63,
...          64,65,66,67,68,69,70,71,72,73,74,75,76,
...          77,78,79,81,82,83,84,85,87,89
...     ]
...     record_split = re.compile(
...              r'([\s]{19})([0-9]{1})([\s]{40})([0-9\s]{2})([0-9\s]{1})([0-9]{1})([0-9]{2})' + 
...              r'([\s]{2})([FM]{1})([0-9]{1})([0-9]{3})([0-9\s]{1})([0-9]{2})([0-9]{2})' + 
...             r'([0-9]{2})([0-9\s]{2})([0-9]{1})([SMWDU]{1})([0-9]{1})([\s]{16})([0-9]{4})' +
...             r'([YNU]{1})([0-9\s]{1})([BCOU]{1})([YNU]{1})([\s]{34})([0-9\s]{1})([0-9\s]{1})' +
...             r'([A-Z0-9\s]{4})([0-9]{3})([\s]{1})([0-9\s]{3})([0-9\s]{3})([0-9\s]{2})([\s]{1})' + 
...             r'([0-9\s]{2})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})' + 
...              r'([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})' + 
...             r'([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})' + 
...              r'([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})' + 
...             r'([A-Z0-9\s]{7})([\s]{36})([A-Z0-9\s]{2})([\s]{1})([A-Z0-9\s]{5})([A-Z0-9\s]{5})' + 
...             r'([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})' + 
...              r'([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})' + 
...              r'([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})' + 
...               r'([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([\s]{1})([0-9\s]{2})([0-9\s]{1})' + 
...             r'([0-9\s]{1})([0-9\s]{1})([0-9\s]{1})([\s]{33})([0-9\s]{3})([0-9\s]{1})([0-9\s]{1})')
...     try:
...              rs = np.array(record_split.split(row))[selected_indices]
...     except:
...             rs = np.array(['-99'] * len(selected_indices))
...     return rs

https://github.com/drabastomek/learningPySpark/blob/master/Chapter02/LearningPySpark_Chapter02.ipynb

GitHub - drabastomek/learningPySpark: Code base for the Learning PySpark book (in preparation)

 

PySpark 배우기의 코드가 올라와있는 깃헙주소이다. 

위 파싱 함수가 깃헙에 있으니 보고 참고하세용 

 

※ 위와 같은 일반적인 파이썬 함수를 선언하면 스파크가 파이썬 인터프리터와 JVM을 지속적으로 스위치해야하기 때문에 애플리케이션이 느려질 수 있다. 때문에 가능하다면 스파크 내장함수를 사용하길 권장한다. 

 

b. 파싱 결과 확인

>>> data_from_file_conv = data_from_file.map(extractInformation)

 

>>> data_from_file_conv.take(1)

 

 

오,, 변환 이전의 데이터랑 비교해보자면

그냥 띄엄띄엄 작성되어있는 것 같았던 데이터 사이에 사실은 공백이 많이 있었구나!