3. 공공데이터포털 데이터 전처리하기(2)

개요

지난 번에 1704 건의 데이터를 전처리하기 앞서 1 건의 데이터를 가져와 내가 원하는 데이터로 변환하는 부분까지 해봤다.
여러 삽질 끝에 원하는 데이터를 만들 수 있었고 1 건의 데이터만 가져와서 테스트해봤기 때문에 이번에는 모든 데이터를 전처리해서
내가 원하는 데이터로 만들어봤다.

 

 

모든 데이터 전처리하기

나는 1704개의 데이터를 일괄 처리하기 힘들 것 같다고 생각이 들었고 한 건의 데이터마다 전처리 후에 추가해주면 되지 않을까 하는 생각이 들었다. 그래서 1건의 데이터를 가져와 전처리하고 추가해주는 소스를 작성하고 결과를 확인해봤다.

 

우선 기본적으로 이전에 적었던 내용을 기본으로 전처리를 진행했다. 크게 틀은 벗어나지 않았던 것 같다.

2. 공공데이터포털 데이터 전처리하기(1)

위의 포스팅 내용을 다시 한 번 정리한다는 느낌으로 적어봐야겠다.

 

json 파일을 읽어서 가져오고

import json
import pandas as pd

# json 파일 읽어 가져오기
with open('/Users/jaynam/workspace/airflow/data/subway_data_20221225.json', 'r') as sw_json:
    sw_json_data = json.load(sw_json)

 

컬럼을 원하는 순서대로 출력하기 위해서 리스트로 만들었다.

sub_list = ['연번', '역번호', '역명', '구분', '호선', '조사일자']

time_list = ['10시00분', '10시30분', 
             '11시00분', '11시30분',
             '12시00분', '12시30분',
             '13시00분', '13시30분',
             '14시00분', '14시30분',
             '15시00분', '15시30분',
             '16시00분', '16시30분',
             '17시00분', '17시30분',
             '18시00분', '18시30분',
             '19시00분', '19시30분',
             '20시00분', '20시30분',
             '21시00분', '21시30분',
             '22시00분', '22시30분',
             '23시00분', '23시30분',          
             '5시30분', 
             '6시00분', '6시30분', 
             '7시00분', '7시30분', 
             '8시00분', '8시30분', 
             '9시00분', '9시30분' ]

 

그리고 이전 글에서 했던 테스트 내용을 함수로 만들었다.
일괄로 어떻게 처리할 수 있는 방법은 없을까? 고민하던 중에 함수로 만들어서 하나씩 변환해서 합치면 되지 않을까? 생각이 들었다.
그래서 이전에 테스트했던 내용을 함수로 만들어서 사용하기로 했다.

def get_sw_ref(sw_df):
    info_df = sw_df[sub_list]
    time_df = sw_df[time_list]
    
    # 행열 바꾸기
    time_rev = time_df.transpose()

    # 컬럼 추가
    time_rev.rename(columns = {0 : '혼잡도'}, inplace = True)
    time_rev['시간'] = time_df.columns.values
    time_rev['연번'] = info_df['연번'][0]

    # 컬럼 순서 변경
    time_rfm = time_rev[['연번', '시간', '혼잡도']]

    # 인덱스 설정
    time_res = time_rfm.set_index('연번')
    
    # 연번을 기준으로 merge
    merge_df = pd.merge(left = info_df , right = time_res, how = "outer", on = "연번")

    return merge_df

이 함수에 대한 내용은 위에서도 말했지만 2. 공공데이터포털 데이터 전처리하기(1) 글에서 확인할 수 있다.

 

 

그리고 변환된 데이터를 저장할 비어있는 Dataframe 을 하나 만들어주었다.
함수를 통해 데이터가 변한되면 비어있는 Dataframe 에 추가되도록 만들어보았다.

sw_result = pd.DataFrame()

for i in range(0, len(sw_json_data['data'])):
    sw_df = pd.json_normalize(sw_json_data['data'][i])
    tmp_df = get_sw_ref(sw_df)
    
    sw_result = pd.concat([sw_result, tmp_df])


sw_result

 

조금 더 자세하게 들여다보면
json 타입으로 되어있는 데이터 중 한 건의 데이터를 가져와 pandas 를 통해 DataFrame 으로 만들어주고

 sw_df = pd.json_normalize(sw_json_data['data'][i])

함수에 넣어서 데이터 변환을 진행해준다.

tmp_df = get_sw_ref(sw_df)

그리고 변환된 데이터를 비어있는 Dataframe 에 추가해준다.

sw_result = pd.concat([sw_result, tmp_df])

 

이렇게 총 1704 건의 데이터를 한 건씩 변환해서 합쳐주었다.

 

그렇게 작성된 최종 소스는 아래와 같다.

import json
import pandas as pd

# json 파일 읽어 가져오기
with open('/Users/jaynam/workspace/airflow/data/subway_data_20221225.json', 'r') as sw_json:
    sw_json_data = json.load(sw_json)


sub_list = ['연번', '역번호', '역명', '구분', '호선', '조사일자']

time_list = ['10시00분', '10시30분', 
             '11시00분', '11시30분',
             '12시00분', '12시30분',
             '13시00분', '13시30분',
             '14시00분', '14시30분',
             '15시00분', '15시30분',
             '16시00분', '16시30분',
             '17시00분', '17시30분',
             '18시00분', '18시30분',
             '19시00분', '19시30분',
             '20시00분', '20시30분',
             '21시00분', '21시30분',
             '22시00분', '22시30분',
             '23시00분', '23시30분',          
             '5시30분', 
             '6시00분', '6시30분', 
             '7시00분', '7시30분', 
             '8시00분', '8시30분', 
             '9시00분', '9시30분' ]

def get_sw_ref(sw_df):
    info_df = sw_df[sub_list]
    time_df = sw_df[time_list]
    
    # 행열 바꾸기
    time_rev = time_df.transpose()

    # 컬럼 추가
    time_rev.rename(columns = {0 : '혼잡도'}, inplace = True)
    time_rev['시간'] = time_df.columns.values
    time_rev['연번'] = info_df['연번'][0]

    # 컬럼 순서 변경
    time_rfm = time_rev[['연번', '시간', '혼잡도']]

    # 인덱스 설정
    time_res = time_rfm.set_index('연번')
    
    # 연번을 기준으로 merge
    merge_df = pd.merge(left = info_df , right = time_res, how = "outer", on = "연번")

    return merge_df
    

sw_result = pd.DataFrame()

for i in range(0, len(sw_json_data['data'])):
    sw_df = pd.json_normalize(sw_json_data['data'][i])
    tmp_df = get_sw_ref(sw_df)
    
    sw_result = pd.concat([sw_result, tmp_df])


sw_result

 

소스를 실행한 결과는 다음과 같이 확인할 수 있었다.

 

 

어떤 데이터를 만들고 싶었는지

마지막으로 내가 어떤 데이터를 만들고싶었는지에 대해서 정리해보려고 한다.

 

원래의 데이터에서는 총 43개의 컬럼으로 되어있었다.

나는 단순하게 시간으로 된 컬럼들을 하나의 컬럼 안에 입력하고 싶었고 그 시간에 맞게 혼잡도를 입력해주고 싶었다.

총 43개의 컬럼 중 시간과 관련된 컬럼은 37개로 되어있었고 그 외의 컬럼은 6개가 남았다.
결과적으로 시간으로 구분되어있는 37개의 컬럼을 2개의 컬럼으로 만들어 입력해주었다.

그 결과 1704 건의 데이터에 각 데이터마다 37개의 시간의 데이터가 생겼기 때문에 총 63048 건의 데이터가 만들어졌다.
컬럼의 수는 8개가 되었다.

1704 * 37 = 63048

 

그렇게 시간에 대한 혼잡도를 시간마다 가져와서 역마다 입력해주었고 만들어진 데이터는 다음과 같았다.

이렇게 만들어진 데이터를 어떻게 활용해볼 수 있는가에 대한 생각은 아직 필요할 것 같다.
그리고 보다 쉽게 활용하기 위한 데이터를 어떻게 만들어볼 수 있을지에 대해서 생각하기 위한 더 많은 시간이 필요할 것 같았다.

 

해야할 일

이제 이렇게 만들어진 파이썬 소스를 Airflow DAG 으로 만들어 PythonOperator 를 통해 실행시켜보려고 한다.
원래 Airflow 를 사용해보고자 데이터를 가져와 전처리 과정을 진행해보았던 것이기 때문에 이제 드디어 실전을 해볼 차례인 것 같다!

 

아, 그리고 데이터를 시각화해서 확인해보는 것도 재미있을 것 같다는 생각이 든다.
2021년도에는 어떤 역이 가장 혼잡했는지 쉽게 확인해볼 수 있지 않을까?