[Flask] Python ORM - SQLAlchemy 대량 데이터 한 번에 삽입 bulk_insert_mappings()

다량의 데이터를 추가/갱신할 때 레거시 코드처럼 Session.add() 메소드는 적절하지 못해보여서 벌크 작업으로 알아보았습니다.

이 상황에서 사용할 수 있는 메소드는 add_all, bulk_save_objects, bulk_insert_mappings / bulk_update_mappings 정도로 나열할 수 있습니다.

또한 SQLAlchemy Core도 포함할 수 있습니다.

---

아래 코드는

for문을 사용해 각 행마다 인스턴스 객체를 생성하고 있습니다. cls()

이 방식은 각 행의 데이터를 데이터베이스에 저장하기 위해 해당 클래스의 인스턴스로 변환하기 위한 것

 

    @classmethod
    def store_data(cls, df) :
        # ▶ 데이터 프레임을 받아 데이터베이스에 저장
        # '_'는 인덱스 값을 무시하고 싶을 때 사용하는 관례적인 표기
        # 'iterrows()' 메서드는 데이터 프레임의 행을 순회할 때 사용하며, 각 행에 대한 인덱스와 행 데이터를 반환합니다. (행 가로줄에 대해 한줄 한줄 나옴)
        for _, row in df.iterrows():
            # row 변수는 현재 순회 중인 행의 데이터를 나타냅니다.

            # to_dict() 메서드는 행 데이터를 딕셔너리 형태로 변환 
            #   └ {'cf_team_id': 202301307, 'cf_team_year': '2023', 'cf_team_mt': '01', 'cf_team_dt': '30', 'cf_team_tm': '심사1팀', 'cf_team_type': 'E', 'cf_team_item': '신', 'cf_team_apct_dtd': 3, 'cf_team_apct_mtd': 4, 'cf_team_cnsl_dtd': 5, ...}
            
            # 변환된 딕셔너리는 ** 연산자를 사용하여 키워드 인자로 전달되며, 이를 통해 데이터베이스 모델의 인스턴스를 생성 
            #   └ cls(cf_team_id=202301307, cf_team_year='2023', cf_team_mt='01', cf_team_dt='30', cf_team_tm='심사1팀', cf_team_type='E', cf_team_item='신', cf_team_apct_dtd=3, cf_team_apct_mtd=4, cf_team_cnsl_dtd=5, ...)
            
            # ▶ 데이터 프레임을 입력으로 받아 데이터베이스에 저장하는 역할
            #    각 행의 데이터를 데이터베이스에 저장하기 위해, 해당 클래스의 인스턴스로 변환하기 위한 것
            #    cls() 클래스메서드에서 인스턴스 생성
            #    for문을 사용해 각 행마다 인스턴스 객체를 생성
            storeData = cls(**row.to_dict())
            db.session.add(storeData)

        db.session.commit()

---

위의 코드에서 약간 변경했는데요.

여러 개의 데이터를 한 번에 삽입하는 경우 bulk_insert_mappings()를 사용할 수 있습니다.

이 방법은 각 행에 대해 인스턴스를 생성하지 않고도 데이터를 삽입하는 데 사용할 수 있는 방법입니다.

 

    # 여러 개의 데이터를 한 번에 삽입하는 경우 bulk_insert_mappings()를 사용할 수 있습니다. 
    # 이 방법은 각 행에 대해 인스턴스를 생성하지 않고도 데이터를 삽입하는 데 사용할 수 있는 방법
    # CFNewSalesStatus 테이블에 삽입되고, 이 과정에서 각 행에 대해 인스턴스를 생성하지 않았습니다. 
    # 이 방식은 인스턴스 생성을 건너뛰고, 데이터베이스 작업을 한 번에 처리하므로 성능 향상이 가능
    @classmethod
    def store_data(df) :
        data_to_insert = []

        # 각 행을 순회하면서 데이터를 data_to_insert 리스트에 추가
        for _, row in df.iterrows():
            data_to_insert.append(row.to_dict())

        # 모든 행을 순회한 후에 bulk_insert_mappings()를 사용해 데이터베이스에 삽입
        # CFNewSalesStatus 모델에 data_to_insert 리스트의 모든 딕셔너리 데이터를 삽입합니다.
        db.session.bulk_insert_mappings(CFNewSalesStatus, data_to_insert)

        # 데이터베이스에 변경 사항을 저장
        db.session.commit()

 

참고 : https://velog.io/@samnaka/sqlalchemy-bulk-insert

---