Dask #4. 대스크 Dataframe과 keras(기초)

1. 대스크 데이터 프레임

 

사용 데이터 : NYC Flights Data ( 뉴욕 3개 지역 공항에서 출발하는 항공편 )

import urllib
import tarfile

# 뉴욕에서 출발하는 항공편 데이터 다운로드 
url = "https://storage.googleapis.com/dask-tutorial-data/nycflights.tar.gz"
filename, headers = urllib.request.urlretrieve(url, './nycflights.tar.gz')

# 압축 해제 코드
with tarfile.open(filename, mode='r:gz') as flights:
    flights.extractall('data/')

 

api로 제공되는 데이터를 다운로드하여줍니다.

만일, 아래와 같은 에러가 난다면 SSL 코드를 실행 후 다운로드하여줍니다.

URLError: <urlopen error [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)>

 

# SSL 에러 발생시 해결방법
import requests
requests.packages.urllib3.disable_warnings()
import ssl

try:
    _create_unverified_https_context = ssl._create_unverified_context
except AttributeError:
    # Legacy Python that doesn't verify HTTPS certificates by default
    pass
else:
    # Handle target environment that doesn't support HTTPS verification
    ssl._create_default_https_context = _create_unverified_https_context

 

데이터가 준비되었다면 대스크의 데이터 프레임으로 불러오겠습니다.

 

import dask.dataframe as dd

df = dd.read_csv('./data/nycflights/*.csv', parse_dates = {'Date': [0, 1, 2]})

 

NYC_flights 데이터를 확인해보면, Year, Month, DayofMonth로 날짜가 구분이 되는 것을 확인하실 수 있는데,

 

Date

1990-01-01,

1990-01-02,

1990-01-03,

1990-01-04,

1990-01-05

 

 

pandas의 기능인 parse_dates = {'Date': [0, 1, 2]}을 활용하여 Date라는 새로운 컬럼에 Datatime 형식으로 불러올 수 있습니다.

 

 

대스크 데이터 프레임의 구조를 확인해보면 위의 이미지와 같이 출력이 됩니다.

npartitions : 파티션 수

int, float, object 등 : 열 데이터 유형

Dask Name : DAG 내부 이름

10 tasks : DAG 노드 수

Pandas의 데이터 프레임과는 다르게 데이터를 불러오지 않으며, 열 데이터의 유형 또한 샘플링을 통해 추정을 한 형태입니다.

 

유형에 대해 조금 더 자세히 보겠습니다.

 

df.tail()

ValueError: Mismatched dtypes found in `pd.read_csv`/`pd.read_table`.

 

대스크의 데이터 프레임으로 불러온 데이터에서 ValueError가 발생합니다. 이는 위에서 설명했듯이 샘플링을 통해 추정한 데이터의 타입이 실제로는 맞지 않아 발생하는 문제로, parquet 형태처럼 데이터 타입까지 같이 저장되는 형식이 아니라면 데이터 유형에 신경을 써주셔야 합니다.

 

df = dd.read_csv('./data/nycflights/*.csv', 
                 parse_dates = {'Date': [0, 1, 2]},
                 dtype={'CRSElapsedTime': 'float64',
                        'TailNum': 'object'}   
                )

 

에러에서 dtype={ --- } 부분을 복사하여 데이터를 다시 불러와주면 데이터의 유형 문제는 해결이 됩니다.

 

2. 대스크와 Keras 연결

 

위에서 데이터 프레임에 대해 가볍게 살펴보았습니다.

Dask의 장점은 Pandas와 동일하게 여러 ML 라이브러리와 연동하여 사용할 수 있는 점입니다.

지금은 아주 간단한 Keras 모델링을 해보도록 하겠습니다.

 

df_train = df[['CRSDepTime',  'CRSArrTime', 'Cancelled']]

 

위에서 불러온 df에서 출발과 도착시간을 통해 취소여부에 대해 예측하는 모델을 만들어 보겠습니다.

 

with ProgressBar() : 
    print(df_train.isnull().sum().compute())

 

총 269,180행의 데이터의 Null 값 여부가 4초 만에 계산되었고, 따로 전처리를 하지 않아도 될 것 같습니다.

모델을 만들어 학습까지 진행해보도록 하겠습니다.

 

from keras.models import Sequential
from keras.layers import Dense

model = Sequential()
# input_dim=df_train.iloc[:, :-1] : Cancelled 컬럼을 제외한 dimension 
model.add(Dense(20, input_dim=df_train.iloc[:, :-1].shape[1], activation='relu'))
model.add(Dense(1,activation='sigmoid'))

model.compile(loss='binary_crossentropy', optimizer='sgd',)

 

model.fit(df_train.iloc[:, :-1].compute(), df_train.iloc[:, -1].compute(),
          batch_size=256,
          epochs=10
         )

 

정말 단순하게 모델링을 해봤습니다. 

빅데이터를 사용할 때에는 generator와 compute()를 사용하여 validation을 적용하여 사용하면 됩니다.

 

아래에는 참고할 링크를 남깁니다.

 

 

Jupyter Notebook Viewer

 

Dask에 대해 설명하는 Youtube와 Jupyter 파일 링크입니다.