[패스트 캠퍼스] 10개 프로젝트로 한 번에 끝내는 MLOps 파이프라인 구현 (실습4)

https://fastcampus.co.kr/courses/233813/clips/

패스트캠퍼스 온라인 강의 - 초격차 패키지 : 10개 프로젝트로 한 번에 끝내는 MLOps 파이프라인 구

Airplane Delay Regression - 항공기가 얼마나 지연되어 도착할 지를 예측하는 모델을 만들어보자. 지연되는 이유를 파악하고 문제점을 해결하거나 대응하여 서비스 품질을 향상시킬 수 있다.

Data Processing 1

1-1 Data Load

dataframe_airlines = pd.read_csv("airlines.csv")
dataframe_flights = pd.read_csv("flights.csv")

외에도 4개의 csv 형식의 데이터가 존재한다. 데이터를 간략하게 파악하고 어떻게 join하여 사용할 지에 대해서 고민해야 한다.

dataframe_airlines.head()
dataframe_flights.info()

등으로 결측치를 파악해볼 수 있다. 
* 데이터를 분석하는 것에 큰 영향을 끼치는지, 어떻게 처리할 것인지에 대한 고민을 하는 시간이다.

각 Column에 대해서 Non-Null Count가 나오고 Dtype은 int64, float64, object등으로 나뉘게 된다.

dataframe_flights.isna().sum()
dataframe_flights.dropna(subset=["arr_delay"]).info()

dataframe_flights 테이블에서 isna().sum()을 하게되면 각 컬럼에 따라 몇 개의 결측치가 존재하는지 알 수 있다. 그 중에서도 arr_delay는 우리가 예측하고자 하는 target column이므로 결측치가 존재하는 경우 dropna 함수를 이용하여 제거해준다. 실습코드에서는 592개의 결측치가 있다.

 

dataframe_airports.info()를 실행하면 faa라는 컬럼값이 존재하는데, Final Arriaval Airport로 예상된다. 같은 맥락으로 공항이름일 것 같은 컬럼이 df_flights에 dest 라는 컬럼이 존재한다. Destination으로 예상된다.

dataframe_airports.faa.unique()
dataframe_flights.dest.unique()
np.intersect1d(dataframe_flights.dest.unique(), dataframe_airports.faa.unique())

df_flight에는 airport 정보가 2개가 있는데, dest와 origin이다. 그 중에서 dest에 기반해서 join할 수 있는지 확인하기 위해서, 두 array간의 intersection이 있는지 확인해주는 numpy 함수를 사용하도록 한다. 그럼 94개정도의 공항이름이 나온다. origin과의 intersection 갯수는 3개의 공항만 나온다.

dataframe_weather = dataframe_weather.drop(["temp", "dewp", "humid","precip","pressure"], axis=1)
dataframe_weather

dataframe의 weather에서 missing value를 많이 포함하고 있는 컬럼들값을 제거해준다.

dataframe_weather.sort_values(["origin", "time_hour"]).groupby("origin").apply(lambda x : x.fillna(method="ffill")) \

날씨 데이터를 origin와 time_hour를 기준으로 정렬 한후에, origin을 바탕으로 그룹화한 뒤에 결측치를 앞뒤값으로 채워준다. Origin은 현재 테이블에서 key 값 (공항이름) 이다.

 

1-3 Data Join

- flight 정보에 airports(공항) 세부 정보를 결합

- flight 정보에 planes(항공기) 세부 정보를 결합

- flight 정보에 출발한 지역의 날씨와 도착한 지역의 날씨 정보를 결합

df_flight_and_airport = pd.merge(df_flights, df_airports, how='left', left_on='dest', right_on='faa')
df_f_a_p = pd.merge(df_flight_and_airport, df_planes, how='left', on='tailnum', suffixes=("_flights","_planes"))

pandas의 merge 함수를 이용해서 df_flights와 df_airports를 dest와 faa를 기준으로 병합해준다. 병합된 새로운 테이블과 항공기 테이블을 tailnum을 기준으로 다시 한번 더 병합해준다. suffixes를 이용해서 양 테이블에 동일한 column명이 있을 때 이름을 다르게해서 조인할 수 있게한다. 첫 번째 테이블 병합에 27개의 컬럼이 있었으나, 두 번째 병합에서 34개의 컬럼으로 증가했다.

 

df_f_a_p["dest_datetime"] = pd.to_datetime(df_f_a_p["time_hour"])
df_f_a_p["dest_datetime"] = df_f_a_p.apply(lambda x : x.dest_datetime + pd.Timedelta(minutes=x.dep_delay) + pd.Timedelta(minutes=x.minute), axis=1)

두 번째 병합된 테이블에 dest_datetime이라는 새로운 컬럼을 생성하고, time_hour 컬럼값을 datetime으로 변환하여 넣어준다. time hour는 2017-10-01 21:00:00 이런형식인데 to_datetime을 쓰면... 어떻게되지? 그 전에는 str에서 그냥 time 포맷으로 변경되는 걸 수도 있겠다. 출발 시간의 날씨 정보에 대해서 join을 세 번째로 진행하기 약간의 처리를 해준다.

time_hour는 분 시간이 제외된 데이터이기 때문에 분을 더해주는데 약간 복잡하고 이해할 필요는 없는 대목이다.

df_f_a_p["time_hour"]가 2017-10-01 21:00:00 이라면, df_f_a_p["dest_datetime"]는 2017-10-02 00:27:00 정도로 출발시간이후에 몇시간이 지난 후 도착지에 몇시에 도착햇는지 알려주는 내용인 것 같다.

df_f_a_p["dest_datetime"] = df_f_a_p["dest_datetime"].dt.strftime("%Y-%m-%d %H:00:00")

날씨 데이터가 시간별로만 나와있기 때문에 뒤에 분을 잘라준다.

df_weather = df_weather.rename(columns={'time_hour': 'dest_datetime' })
df_final = pd.merge(df_f_a_p, df_weather, how="left", on=["origin","dest_datetime"], suffixes=("", "_weather"))

이후 귀찮아서 dataframe으로 안고치고 df 코드복사. join을 위해서 time hour값을 dest_datetime으로 이름 변경해준다. left join에서 on하나로 편하게 할라고? left_on, right_on안하고? 그래서 하는건가? 세번째 병합으로 날씨 데이터를 더해주는데 origin과 dest_datetime을 기준으로 해주고 suffixes에서 동일한 컬럼명을 가진 weather에 변화를 준다. "" 는뭔데?

print("duplicated :", len(df_final[df_final.duplicated()]))

중복된 데이터가 있는지 확인하는 과정을 통해서 join에서 문제가 있었는지, data collection 및 prep단계에서 문제가 있었는지 double check한다.

2. Data EDA

df_final.info()
df_final.isna().sum()

# categorical data와 numerical data를 나눈다.
for column_name in list(df_final.columns):
    print(column_name, type(df_final[column_name][0]), df_final[column_name].unique())
    
df_final.select_dtypes(include=['int64']).columns
df_final.select_dtypes(include=['float64']).columns

unique값이 출력되며 대충 어떤 컬럼이 분류형인지, 수치형인지 생각해본다. Int64 type의 컬럼이 categorical한 특성응ㄹ 보이기도 해서 휴리스틱하게 판단한다. 

Int64 :

Index(['year_flights', 'month', 'day', 'sched_dep_time', 'sched_arr_time'], dtype='object')

 

Float64:

Index(['dep_time', 'dep_delay', 'arr_time', 'arr_delay', 'flight', 'air_time',
       'distance', 'hour', 'minute', 'lat', 'lon', 'alt', 'tz', 'tzone',
       'year_planes', 'engines', 'seats', 'year', 'month_weather',
       'day_weather', 'hour_weather', 'wind_dir', 'wind_speed', 'wind_gust',
       'visib'],
      dtype='object')

 

 

과정에서 dep_time과 arr_time은 시간을 의미하고, hour과 minute 데이터가 datetime형태 (%H%M)이기 때문에 변환이 필요하다 것과, wind_dir도 categorical한 특성으로 간주할 수 있다는 인사인트를 대강 얻는다.

df_final["dep_time"] = df_final["dep_time"].apply(lambda x: '{0:04d}'.format(int(x)))
df_final["arr_time"] = df_final["arr_time"].apply(lambda x: '{0:04d}'.format(int(x)))
df_final["wind_dir"] = df_final["wind_dir"].astype("str")

dep_time과 arr_time 변환. 0.04d는 ... 뭘 의미하는거냐?

list_categorical_columns = list(df_final.select_dtypes(include=['int64', 'object']).columns)
list_numeric_columns = list(df_final.select_dtypes(include=['float64']).columns)
target_column = "arr_delay"

int64와 object를 카테고리컬에 넣고 float64를 뉴메릭에 넣은 후에 갯수 확인.

list_categorical_columns = list(df_final.select_dtypes(include=['int64', 'object']).columns)
list_numeric_columns = list(df_final.select_dtypes(include=['float64']).columns)
target_column = "arr_delay"
print(len(df_final.columns))
print(len(list_categorical_columns))
print(len(list_numeric_columns))

 

Dependent Data Exploration

sns.histplot(data=df_final, x="arr_delay")
print(stats.skew(df_final["arr_delay"]))
print(stats.kurtosis(df_final["arr_delay"]))
sns.boxplot(data=df_final, x="arr_delay")
sns.boxplot(data=df_final[df_final.arr_delay < 100], x="arr_delay")
df_final["arr_delay"].describe()

종속변수에 대해서 histplot을 시각화하고, skewness와 왜곡정도를 확인한다. boxplot을 통해서 전반적으로 25~50%의 종속변수가 어느 범위에 존재하는지 확인하고, 그래프를 보기가 어렵다면 arr_delay 시간을 100이하로 줄여서 가시성을 높인다.

Independent Data Exploration

df_final.describe().to_csv(path_lecture + "data/describe.csv")

컬럼의 갯수가 많은 경우, describe내용을 csv로 다운로드 받은 후에 천천히 살펴보며 정보를 찾아볼 수 있다.

df_final.apply(lambda x : x.unique())

어떤 컬럼이 불필요한지 고민해서 정해볼 수가 있다.

year_flights : 2017년 하나만 있어서 의미가 없다.

dep_time : 실제 출발시간인 것 같다. 실제로 언제 출발했는지는 큰 의미가 있을까? 그럴 수 있다. 기상환경이 시간에 따라 달라질 수 있기 때문이다.

sched_dep_time : 출발 예정시간에 따라 실제 출발시간이 변할 수 있다. 오히려 sched_dep_time을 남기고 dep_time을 제거하는 게 효율적일수도 있다. 

arr_time은 실제 도착시간인데 의미가 없을 것 같다.

sched_arr_time : 도착 예정시간인데, 예정시간에 따라서도 비행기의 착륙시간이 지연될 수 있다.

arr_delay : 타겟값

carrier : 운행사에 따라 달라질 수 있다.

오히려 위도.경도 값이 무의미할 것 같다.

month/day/hour weather는 제거해주도록 한다.

Categorical column data exploration

list_categorical_columns
sns.boxplot(data=df_final, x="arr_delay", y="origin", hue="origin")
sns.boxplot(data=df_final[df_final.arr_delay < 100], x="arr_delay", y="origin", hue="origin")
sns.boxplot(data=df_final, x="arr_delay", y="engine", hue="engine")
sns.boxplot(data=df_final, x="distance", y="engine", hue="engine")
sns.scatterplot(data=df_final, x="month", y="arr_delay")
sns.scatterplot(data=df_final, x="carrier", y="arr_delay")
plt.figure(figsize=(10,10))
sns.boxplot(data=df_final, x="arr_delay", y="carrier")
plt.figure(figsize=(10,10))
sns.boxplot(data=df_final[df_final.arr_delay < 200], x="arr_delay", y="carrier")

출발지/ 엔진과의 박스플롯. 엔진과 distance간의 박스플롯. month와 지연시간과의 스캐터플롯, 항공사와 지연시간과의 스캐터플롯. 운항사와 도착 지연시간과의 구별 등.

 

##### Categorical Data와 Numerical Data 검정
*   두 데이터의 연관성을 검정하는 방법에는 모수적 방법과 비모수적 방법으로 나뉨
*   모수적 방법 : ANOVA
*   비모수적 방법 : Kruskall-wallis
*   모수적 방법은 다양한 조건을 만족해야 하기 때문에 적용이 어려울 수 있으나, 정확도는 비모수적 방법에 비해 높음

# Normality - QQplot, Shapiro Wilk test
stats.probplot(df_final[target_column], dist=stats.norm, plot=plt)
# Non-parametric - Kruskall-Wallis test
list_meaningful_column_by_kruskall = []

for column_name in list_categorical_columns:
  list_by_value = []
  for value in df_final[column_name].dropna().unique():
      df_tmp = df_final[df_final[column_name] == value][target_column].dropna()
      list_by_value.append(np.array(df_tmp))
  statistic, pvalue = kruskal(*list_by_value)
  if pvalue <= 0.05:
    list_meaningful_column_by_kruskall.append(column_name)
  print(column_name, ", ",statistic,", ", pvalue)
print("all categorical columns : ", len(list_categorical_columns))
print("selected columns by kruskal : ", len(list_meaningful_column_by_kruskall), list_meaningful_column_by_kruskall)

Numeric (Continuous) column Data Exploration

sns.pairplot(data=df_final.loc[:,list_numeric_columns])
df_corr = df_final.loc[:,list_numeric_columns].corr()
plt.figure(figsize=(10,10))
sns.heatmap(df_corr, annot=True)
index_corr_over_90 = np.where((abs(df_corr) > 0.90) & (df_corr != 1))
index_corr_over_90
len_corr_over_90 = len(index_corr_over_90[0])
left_columns = df_corr.columns[index_corr_over_90[0]]
right_columns = df_corr.columns[index_corr_over_90[1]]
for index in range(len_corr_over_90):
  print(left_columns[index], "<->", right_columns[index])

list_removed_by_correlation = ["air_time", "lon", "wind_speed"]

numeric columns들 간의 pairplot을 찍어내고, 각각의 피쳐들간의 상관관계를 알아본다. 히트맵을 찍어서 시각적으로 나타낼 수도 있다. correlation이 0.9 이상이거나 1보다 작은 피쳐를 찾아내고 서로 짝지어본다. 그 결과는

dep_delay <-> arr_delay
arr_delay <-> dep_delay
air_time <-> distance
air_time <-> lon
distance <-> air_time
distance <-> lon
lon <-> air_time
lon <-> distance
wind_speed <-> wind_gust
wind_gust <-> wind_speed

air_time과 lon, wind_speed를 제거하기로 한다. faa, origin, dest도 join할 때 사용된 키값으로, 다른 데이터의 key와 동일하기 때문에 제거해준다. 

 

 

 

 

 

 

 

 

 

 

 

1. df_flights.info() 결과는 30만건이 아니라 5만건. 실습 녹화당시 이후에 데이터셋 변경이 있던 것으로 보임. 그 결과 df_flights.isna().sum()에서 나타나는 결측치의 갯수와 비율이 완벽하게 맞지않음. 명확하게는 Flight, origin, dest, distance, hour, minute, time_hour에서 1개씩의 결측치가 존재함.

2. 그 결과 np.intersect1d(df_flights.dest.unique(), df_airports.faa.unique())

np.intersect1d(df_flights.dest.unique(), df_airports.faa.unique())

에서 str과 float을 비교할 수 없다는 에러 메시지가 발생. 해당 셀을 실행하기 전에 df_flights = df_flights.dropna(subset=["dest"]).copy() 을 실행해서 'dest'에 있는 Nan값을 제거해야함.

 

3. 날씨 결측치는 앞뒤 날씨가 큰 영향을 미치므로

df_weather.sort_values(["origin", "time_hour"]).groupby("origin").apply(lambda x : x.fillna(method="ffill")) \
                                                    .groupby("origin").apply(lambda x : x.fillna(method="bfill")).info()