Bigquery load error (GenericGBQException)_(동시접근, parallel 작업시)

GCP를 활용하다 보면 Bigquery의 힘이 정말 대단하다.

빠르고 대용량을 아주...꽤나 잘 처리한다.

주로 빅데이터를 분석하면서 bigquery에서 데이터를 불러와 잘짜여진 알고리즘에 따라

분석된 결과를 다시 Google BQ에 올려 언제든 결과를 확인하고 싶을때가 많다.

 

그리고 그렇게 일을 처리해야 GCP를 이용해 업무를 효율적으로 진행할 수 있다.

 

필자는 데이터 분석을 할때 parallel 한 처리를 하기 때문에

parallel로 처리한 결과들을 비동기식으로 bq에 적재하는 경우가 많다.

 

import multiprocessing
### bigquery load error 일부러 내기
def bq_load(df):
    ....
    result = 
    result.to_gbq(destination_table=figure_destination_table, project_id=project_id, if_exists='append', credentials=credentials, verbose=False)
    
final_df_split = np.array_split(final_df[col], len(list(final_df[col])), axis=1)
pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
result = pool.map(bq_load,final_df_split)
pool.close()
pool.join()

이때 주로 나타날수 있는 error 가 GenericGBQException 에러다.

 

이는 각 process가 처리한 결과를 bq에 있는 한개의 table에 append를 이용하여 적재할 때,

다중 process가 한 곳에 load 하는 job이 집중되며 나오는 현상이다.

 

이를 해결하기 위한 보수적인 방법은

1. 실패하면 retry를 시키던가

import multiprocessing
from retry import retry

def GenericGBQException()
	...
    
@retry(exceptions=GenericGBQException, tries=3, delay=2)
def bq_load(df):
    ....
    result = 
    result.to_gbq(destination_table=figure_destination_table, project_id=project_id, if_exists='append', credentials=credentials, verbose=False)
    
final_df_split = np.array_split(final_df[col], len(list(final_df[col])), axis=1)
pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
result = pool.map(bq_load,final_df_split)
pool.close()
pool.join()

2. 각 process마다 sleep을 random하게 걸던가? 이다.

import multiprocessing
from random import randint
import time

### bigquery load error 일부러 내기
def bq_load(df):
    ....
    result =
    time.sleep(randint(0,5))
    result.to_gbq(destination_table=figure_destination_table, project_id=project_id, if_exists='append', credentials=credentials, verbose=False)
    
final_df_split = np.array_split(final_df[col], len(list(final_df[col])), axis=1)
pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
result = pool.map(bq_load,final_df_split)
pool.close()
pool.join()

 

위 방법들은...일련의 잔머리로 적용한 것이지만..

사실 위의 방법이 작업시간을 줄이려는 의도를 벗어나긴 한다.

 

그래서..위 에러를 벗어나서 그냥 결과를 다 map을 이용해 모은 다음 한방에 gbq에 로드하는 방식으로 바꾸긴 했지만...

 

조금 더 작업속도를 줄일 다른 방법들은 차근차근 살펴봐야겠다.