postgres table join을 이용한 data update

pandas dataframe을 이용하여 db 업데이트 작업을 하다 문제가 생겼다.

id	primaykey
tracking_number	int
created_at	date

위와 같은 fulfillment table 에서 특정 id 값들의 created_at값을 일괄적으로 변경한다 했을때 targets에 해당하는 id 값들을 transaction 한번에 모두 처리할수 있다.

engine = LOCAL_DB
targets = (1,7,15,3000,4000,...13000)

update_query = f'''
UPDATE fulfillment
SET
created = now()
WHERE
id in {targets}
'''

with engine.connect() as conn:
    conn.execute(update_query)

하지만 id값마다 다른 tracking_number를 넣는다고 할때 루프를 태우는 방법외엔 따로 없었다.

만약 insert라면 dataframe 자체를 to_sql메서드를 사용하여 일괄적으로 insert하기 때문에 문제가 없다.

insert_dataframe # fulfillment테이블과 동일한 포맷의 target dataframe이 있다고 가정한다.
insert_dataframe.to_sql('fulfillment', con=ENGINE_LOCAL, if_exists='append', index=False, method='multi', chunksize=50000)

여기서 문제가 발생하였다. to_sql 메서드를 사용하여 특정 키값에 해당하는 필드를 업데이트 하고 중복된다면 생성하지 않고 넘어가야했는데 to_sql메서드에서는 이러한 기능을 제공하지 않는걸로 확인이 됐다(물론 제가 못찾았을수도...).

물론 for루프를 태워서 해결은 가능하다.

insert_dataframe
with engine.connect() as conn:
    for i in insert_dataframe.index:
        update_query = f'''
        UPDATE fulfillment
        SET
        tracking_number insert_dataframe['tracking_number'][i]
        WHERE
        id = insert_dataframe['id'][i]
        '''

        conn.execute(update_query)

위와 같이 단건으로 트랜잭션을 발생시키고 네트워크를 태우는 성능 최저의 방식을 사용한다했을때 데이터 1만개 기준 1시간이 넘어가는 성능을 보여줬다. 물론 안정성은 보장되고 데이터의 수가 적다면 위의 인덱스 하나하나 처리하는 방식이 맞다고 할수 있다.

하지만, 데이터 20만개정도를 업데이트한다고 했을때는 말도 안되는 저성능을 보여준다.

 

이때 사용한것이 temp 테이블을 이용한 join 업데이트다. 물론, 애플리케이션 단에서 모두 처리할 수 있는 방법이 있을것이라 생각하는데 찾아내지는 못했다. 여기서 조건은 업데이트할 테이블이 속해있는 database에 temp 테이블이 있어야한다. 방법은 아래와같다.

 

1. fulfillment_table과 동일한 포맷인 fulfillment_temp_table을 create한다.

2. fulfillment_table에 target데이터인 insert_dataframe을 to_sql메서드를 통해 insert한다.

3. left join을 통해 일괄적으로 fulfillment_table_tracking_number를 update한다.

insert_dataframe # to_sql로 insert할 target_data

insert_dataframe.to_sql('fulfillment_temp', con=ENGINE_LOCAL, if_exists='append', index=False, method='multi', chunksize=50000)

update_query = '''
UPDATE
fulfillmet as a
SET
tracking_number = b.tracking_number,
created_at = b.created_at,
FROM
fulfillment_temp as b 
WHERE
a.id = b.id
'''
with engine.connect() as conn:
    conn.execute(update_query)

위와 같은 방식으로 join을 이용한 대량의 update를 할수 있다.

 

bulk_update 방식으로 업데이트 데이터들의 query를 일일이 생성하여 대량 업데이트하는 방식도 있지만 이방식이 가장 간단했다고 생각이 든다.