Introduction
다른 데이터 Source에서 Pandas의 Dataframe으로 데이터를 Extract(추출)한 후에 Destination으로 데이터를 Load(적재)하는 경우에 데이터의 type을 casting해줘야 하는 경우가 많이 발생합니다. 타입에 관한 문제들을 겪다보면 단일 데이터베이스에서 ORM으로 DB에 CRUD를 하는 것이 얼마나 생산성이 높은지 느낄 수 있습니다.
특히, typing이 되어있는 데이터 소스가 아닌 경우(Web Page Crawling)하는 경우에 특히 이런 Needs들이 있습니다.
이런 경우에 Type casting이 필요한데 기본적인 Type casting과 나름의 시행착오를 겪은 부분들을 정리하려고 합니다. 특히, BigQuery에 Data를 load하는 경우에 type casting에 대한 소소한 Tip들도 적어 보겠습니다.
Type Casting에 앞서 가장 먼저 확인 해야 하는 부분은 DataFrame의 데이터의 형태를 확인해야 합니다.
조금 더 구체적인 데이터와 같이 살펴보기 위해서 Naver Finance에서 Sample 데이터를 가져옵니다.
KOSPI 200 첫 페이지 회사 리스트 가져오는 코드
# Python 3.8.8
from re import search
import requests
from bs4 import BeautifulSoup
import pandas as pd
url = 'http://finance.naver.com//sise/entryJongmok.naver?page=1'
session = requests.Session()
response = session.get(url)
html = BeautifulSoup(response.text, 'html.parser')
# column 가져오기
columns = []
for tag in html.select('table th'):
columns.append(tag.text)
# row 가져오기
def clean_up_string(data: str) -> str:
match = search('[\S]+', data)
if match:
result = match.group(0)
else:
result = data
result = result.replace(',' ,'')
return result
rows = []
for _row in html.select('tr'):
row = []
for tag in _row.select('td'):
row.append(clean_up_string(tag.text))
if len(row) == len(columns):
rows.append(row)
df = pd.DataFrame(rows, columns=columns)
1. 데이터 살펴보고 작업 계획 세우기
데이터를 casting하기 전에 현재 데이터가 어떤 타입인지 살펴봐야 합니다. 가끔 마음이 급해서 그냥 데이터를 Extract하고 바로 Load하려고 하다가 에러를 하나씩 해결하는 경우도 있는데 대부분 코드를 고치는데 더 오랜 시간을 필요로 합니다. 현재 데이터 타입이 어떻게 되어 있고 Destination의 Schema를 확인하고 어떻게 casting해서 Load할지 천천히 살펴보는것이 중요합니다.
미리 준비한 코드를 이용해 Dataframe의 Info() 함수와 head() 함수를 통해 데이터를 확인합니다.
df.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 10 entries, 0 to 9
Data columns (total 7 columns):
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 종목별 10 non-null object
1 현재가 10 non-null object
2 전일비 10 non-null object
3 등락률 10 non-null object
4 거래량 10 non-null object
5 거래대금(백만) 10 non-null object
6 시가총액(억) 10 non-null object
dtypes: object(7)
memory usage: 688.0+ bytes
df.head()
종목별 현재가 전일비 등락률 거래량 거래대금(백만) 시가총액(억)
0 삼성전자 76900 1300 -1.66% 9091202 701044 4590763
1 SK하이닉스 120500 3000 -2.43% 2436028 293151 877243
2 NAVER 395000 4500 -1.13% 219062 86481 648840
3 삼성바이오로직스 901000 15000 -1.64% 39219 35511 596147
4 카카오 122500 0 0.00% 957002 116534 54608
가장 중점적으로 확인해야 할 부분은 두 가지입니다.
- DataType
- Null 데이터 유무 -> null 데이터 유무에 따라서 casting할 때 에러가 발생할 수 있습니다. (int로 캐스팅 하는 경우)
우리가 이 데이터를 Load할 bigquery table schema는 아래와 같다고 가정합니다.
# biguqery schema list로 예시를 위해서 가져왔습니다.
schema = [
SchemaField("company_name", "STRING", mode="REQUIRED"),
SchemaField("current_price", "FlOAT"),
SchemaField("amount_changes", "FlOAT"),
SchemaField("ratio_changes", "FlOAT"),
SchemaField("trading_volume.", "INT64"),
SchemaField("trading_value", "FlOAT"),
SchemaField("market_cap", "FlOAT"),
]위와 같은 데이터로 변환하기 위해서는
- 현재가 -> float
- 전일비 -> florat
- 등략률 -> florat
- 거래량 -> integer
- 거래대금 (백만) -> float
- 시가총액 (억) -> float
을 변환해야 합니다. 참고로 NaN(Not a number)를 Float로 casting할 때는 상관없지만, Integer로 변환하는 경우가 에러가 발생합니다. 예를 위해서 삼성전자의 거래량을 NaN으로 변경합니다.
from numpy import NaN
df.loc[0:0, ['거래량']] = NaN
df.head()
2. Type Casting 하기
이제 본격적으로 type casting을 시작합니다.
Type Casting에 앞서 type별로 casting하는 방법을 살펴봅니다.
# int32
df['column_name'].astype('int32')
# float32
df['column_name'].astype('float32')
# object
df['column_name'].astype('object')
# category
df['column_name'].astype('category')
# datetime or datetime64[ns]
df.loc[:, 'column_name'] = pd.to_datetime(df['column_name'])
이제 위에 정리한 내용처럼 type을 casting합니다.
- 현재가 -> float
- 전일비 -> florat
- 등략률 -> florat
- 거래량 -> integer
- 거래대금 (백만) -> float
- 시가총액 (억) -> float
아래와 같이 현재가, 전일비, 등락률, 거래대금(백만), 시가총액을 변경합니다.
df.loc[:, '현재가'] = df['현재가'].astype('float')
# float로 type이 변경된 것을 확인 할 수 있습니다.
df['현재가']
0 76900.0
1 120500.0
2 395000.0
3 901000.0
4 122500.0
5 740000.0
6 709000.0
7 207500.0
8 85400.0
9 63400.0
Name: 현재가, dtype: float64
거래량을 integer로 변경합니다. 아래와 같이 NaN을 integer로 바꿀 수 없다는 에러가 발생합니다.
df.loc[:, '거래량'] = df['거래량'].astype('int64')
df['거래량']
---------------------------------------------------------------------------
ValueError Traceback (most recent call last)
/var/folders/_k/w0m1fxfx2nngqbf5r4sp65lr0000gn/T/ipykernel_77441/4196606652.py in <module>
----> 1 df.loc[:, '거래량'] = df['거래량'].astype('int64')
2 df['거래량']
.....
ValueError: cannot convert float NaN to integer1. null handling 후에 casting 해준다. 입력되지 않음을 -1 표시.
df.fillna({'거래량': -1}, inplace=True)
df.loc[:, '거래량'] = df['거래량'].astype('int64')
df['거래량']
0 -1
1 2436028
2 219062
3 39219
4 957002
5 243786
6 183129
7 442537
8 1685784
9 2385539
Name: 거래량, dtype: int64
2. BigQuery의 loadjob을 schmea와 함께 사용하는 경우는 float casting을 한 후에 load할 때 pyarrow에서 integer로 casting을 해주기 때문에 float로 casting 해줘도 됩니다. 저는 BigQuery에 data를 Load하는 경우는 이 방식을 더 선호합니다.
df.loc[:, '거래량'] = df['거래량'].astype('float')
job = bq_client.load_table_from_dataframe(
dataframe=df,
destination=table_name,
job_config=bigquery.LoadJobConfig(schema=schema, write_disposition="WRITE_TRUNCATE")
)
job.result()
마지막으로 object를 datetime으로 변경하는 부분만 같이 살펴보겠습니다.
# 예시를 위해서 오늘 날짜로 주식 데이터의 날짜 설정
df['record_date'] = '2021-12-12'
# type이 object
df['record_date']
0 2021-12-12
1 2021-12-12
2 2021-12-12
3 2021-12-12
4 2021-12-12
5 2021-12-12
6 2021-12-12
7 2021-12-12
8 2021-12-12
9 2021-12-12
Name: record_date, dtype: object
# type을 datetime64로 변경.
df.loc[:, 'record_date'] = pd.to_datetime(df['record_date'])
df['record_date']
0 2021-12-12
1 2021-12-12
2 2021-12-12
3 2021-12-12
4 2021-12-12
5 2021-12-12
6 2021-12-12
7 2021-12-12
8 2021-12-12
9 2021-12-12
Name: record_date, dtype: datetime64[ns]
마무리
이것으로 Pandas에서 type casting을 간단하게 정리해봤습니다. 데이터를 Source에서 Destination으로 옮길 때 Type을 casting할 일이 생기는데 이 때 데이터를 잘 확인하고 효율적으로 casting하는 것이 중요합니다. 다음에는 저의 공부를 위해서 casting하는 API들의 performance나 best practice에 대해서도 한 번 정리해 보면 좋을 것 같습니다.
읽어주셔서 감사합니다. 좋은 연말 되시길!
'Computer Engineering > Data Engineering' 카테고리의 다른 글
| Airflow Taskflow로 DAG refactoring하기 (0) | 2022.11.14 |
|---|---|
| Pandas NaN이란 그리고 None 차이 (0) | 2022.11.13 |
| Airflow Sensor 정리 (feat. S3 Sensor) (0) | 2022.07.12 |
| Airflow k8s 로컬 개발환경 셋팅 (2) | 2022.07.08 |
| IntelliJ로 Spark 개발 환경 구축하기 (0) | 2022.07.01 |