[pandas] 메모리 관리를 위한 대용량 데이터프레임 다루기

python에서 pandas로 대용량 데이터를 처리하면서 속도가 매우 느려서 전처리에 어려움을 겪었다.
불러오려는 데이터의 크기는 3~5GB 였으며, 파이프( | )로 분리된 csv파일이다.
이를 기준으로 시도했던 방안을 정리했다.

---

1. 일부 row를 통해 컬럼과 형태 확인하기

import pandas as pd
import time
file_path = '파일명.csv'
df = pd.read_csv(file_path, delimiter='|',nrows=3)

• 데이터의 일부 row만 불러오면, 데이터 형태 확인 정도는 빠르게 가능하다.

2. 적절한 라이브러리를 이용해 DataFrame으로 불러오기

ⓐ pandas의 read_csv

# 불러오기
start_time = time.time() # 시작시간
pd_df = pd.read_csv(file_path, delimiter='|', usecols=['BASE_YM','기업고유번호','LCDONG_CD','BLD_NO_A','BLD_NO_B','BY1PT_USAGE'])
print("read time :", time.time() - start_time) # 현재시간 - 시작시간

--------------------
> read time : 362.4267168045044

# 연산
start_time = time.time()  # 시작시간
pd_df = pd_df.drop_duplicates(subset=['BASE_YM', '기업고유번호', 'BY1PT_USAGE']) # 중복 행 제거
pd_df_sum = pd_df.groupby('BASE_YM')['BY1PT_USAGE'].sum().reset_index() # 월별 집계 후 concat
print("calc time :", time.time() - start_time)
#del pd_df

--------------------
> calc time : 32.06459999084473
> total time : 394.4913167953491

• 일반적으로 사용하는 방식
• 데이터프레임으로 불러오는 속도가 느리다.

 

ⓑ pyarrow

import pyarrow.csv as pv
import pyarrow.dataset as ds
# 불러오기
start_time = time.time()  # 시작시간
selected_columns = ['BASE_YM', '기업고유번호', 'LCDONG_CD', 'BLD_NO_A', 'BLD_NO_B', 'BY1PT_USAGE'] # 불러올 컬럼 지정
read_options = pv.ReadOptions(use_threads=True) # 멀티스레딩사용
parse_options = pv.ParseOptions(delimiter='|') # 구분자설정
convert_options = pv.ConvertOptions(include_columns=selected_columns)
pyarrow_table = pv.read_csv(file_path, read_options=read_options, parse_options=parse_options, convert_options=convert_options)
pyarrow_df = pyarrow_table.to_pandas() # 데이터프레임으로 변환
print("read time :", time.time() - start_time) # 현재시간 - 시작시간
#del pyarrow_table, pyarrow_df

--------------------
> read time : 334.4773008823395

• Apache Arrow는 인메모리 분석을 위한 개발 플랫폼이다.
• Columnar In-Memory 포맷(메모리상에서 컬럼구조로 데이터 정의)과 pandas, spark 등 여러 시스템에서 같은 메모리 포맷을 공유한다.
  (Arrow는 C++구현)
  → 메모리 문제는 해결해주지 못한다.
  

 

ⓒ dask의 read_csv

import dask.dataframe as dd
# 불러오기
start_time = time.time()  # 시작시간
dask_df = dd.read_csv(file_path, sep='|', usecols=['BASE_YM','기업고유번호", 'LCDONG_CD','BLD_NO_A','BLD_NO_B','BY1PT_USAGE'])
print("read time :", time.time() - start_time) # 현재시간 - 시작시간

--------------------
> read time : 0.14893460273742676

# 연산
start_time = time.time()
dask_df = dask_df.drop_duplicates(subset=['BASE_YM', '기업고유번호', 'BY1PT_USAGE'])# 중복 행 제거
dask_df_sum = dask_df.groupby('BASE_YM')['BY1PT_USAGE'].sum().reset_index()# 월별 집계 후 concat
print("calc time :", time.time() - start_time)
#del dask_df

--------------------
> calc time : 0.019330978393554688

# COMPUTE
from dask.diagnostics import ProgressBar
from time import sleep
pbar = ProgressBar()
pbar.register()
#%%time
start_time = time.time()
dask_df_sum = dask_df_sum.compute()
print("compute time :", time.time() - start_time)

--------------------
> compute time : 360.67797470092773

• 가상 데이터프레임을 사용해서 대규모 데이터셋을 병렬로 처리한다.
  데이터를 청크(chunk)로 분할 > 청크를 처리하는 작업을 정의하고, 의존성을 그래프로 표현 > 작업 스케줄링 > 여러개의 작업을 병렬로 실행(의존성 고려)
• 디스크에 저장된 데이터를 분산처리하는 방식. 지연 연산을 사용하기에 실제 연산을 최적화한다.
• 메모리가 감당 가능한 수준의 연산이라면, 메모리와 디스크의 속도 차이 등으로 인해 Pandas가 유리하다.
• 실제로 데이터를 불러오고 연산하는 시간은 1초 이하로 매우 빨랐다. 하지만 pandas dataframe으로 변환하려면 compute()함수를 사용해야하는데, 이 작업시간이 오래 걸린다.
  → compute()로 연산 결과를 확인하는 시간이 여전히 오래 걸린다.
  

 

3. 일부 컬럼만 불러오기/남기기

# 전체 
start_time = time.time()
pd_df = pd.read_csv(file_path, delimiter='|',nrows=100000)
print("전체 time :", time.time() - start_time) # 현재시간 - 시작시간
del pd_df
# 일부 컬럼
start_time = time.time()
pd_df = pd.read_csv(file_path, delimiter='|', usecols=['BASE_YM','LCDONG_CD','BLD_NO_A','BLD_NO_B','BY1PT_USAGE','GAS_USAGE'],nrows=100000)
print("일부 time :", time.time() - start_time) # 현재시간 - 시작시간
del pd_df

--------------------
> 전체 time : 1.6558094024658203
  일부 time : 0.1175544261932373

• 전체가 아닌 일부만 불러오는 것이 속도면에서 10배 이상 빠르다.

df = df[['BASE_YM','LCDONG_CD','BLD_NO_A','BLD_NO_B','BY1PT_USAGE']]

• 전처리 과정에서도 데이터를 불러올 때와 마찬가지로 필요한 컬럼만 남겨 메모리를 절약하자.

4. 반복문 최적화

for문은 while문보다는 빠르지만, 여전히 매우 느리다.
for문 내부와 pandas는 python언어로 동작하기때문에 속도가 매우 느리다.
반복문을 써야한다면 for문 내부는 간단하게 하고, 최대한 외부에서 처리하는 방법이 빠르다.

 

5. 불필요한 객체 삭제

del df

사용하지 않는 변수는 del을 이용하여 삭제한다.
객체를 바로 지우지는 않지만, 변수를 지우고 다른 변수가 참조하고있지 않다면 객체는 지워지게된다.

 

6. 메소드 사용 X, 내장 함수와 라이브러리 사용 O

메소드(.)를 사용하면 객체에서 메소드를 열거하고 일치하는 메소드를 찾는 연산을 수행한다.
따라서 많은 시간이 소요되므로 메소드 대신 내장함수와 라이브러리를 적극적으로 사용하는 것이 시간 절약에 도움이 된다.

---

 

.
.
.

Spark 사용의 필요성을 느꼈다..