빙응의 공부 블로그

[+/KM]JDBC Batch로 CSV 데이터 저장 최적화 본문

Project/PlusKM

[+/KM]JDBC Batch로 CSV 데이터 저장 최적화

빙응이 2025. 2. 18. 23:21

 

📝배경

저번 포스팅 에서는 CSV 파일을 이용해 프로젝트 데이터를 불러오는 과정을 진행했습니다. 또한, MySQL 데이터를 CSV로 변환한 후 MongoDB로 복사하는 작업도 수행했습니다.

이 프로젝트의 특성상 데이터는 약 10만 건으로 큰 변화가 없으며, 서버 실행 시 갱신됩니다. 원래는 한 달에 한 번 CSV 파일이 갱신되면, 서버를 재시작해 데이터를 반영하는 방식을 계획했습니다. 

 

[Place] 데이터 동기화 완료 - 추가/수정: 100676, 삭제: 0, 시간: 172223ms // 172초 거의 3분

📝MySQL 데이터 추가 최적화하기 

우선 MySQL 데이터 추가 및 갱신 로직을 최적화하겠습니다.
데이터를 추가할 때는 기존 DB와 비교하여 중복 여부를 확인하고, 중복 데이터는 삭제, 갱신된 데이터는 업데이트합니다.

    private void updateDatabase(Map<String, Place> csvDataMap) {
        // 기존 데이터 조회
        List<Place> existingPlaces = placeRepository.findAll();

        // 기존 데이터 맵 생성
        Map<String, Place> existingPlacesMap = existingPlaces.stream()
            .collect(Collectors.toMap(
                place -> generateUniqueKey(place.getName(), place.getAddress()),
                place -> place
            ));

        // 추가/수정 대상 찾기
        List<Place> placesToSave = csvDataMap.entrySet().stream()
            .filter(entry -> !existingPlacesMap.containsKey(entry.getKey()) || isUpdated(entry.getKey(), entry.getValue(), existingPlacesMap))
            .map(Map.Entry::getValue)
            .collect(Collectors.toList());

        // 삭제 대상 찾기
        List<Place> placesToDelete = existingPlaces.stream()
            .filter(place -> !csvDataMap.containsKey(generateUniqueKey(place.getName(), place.getAddress())))
            .collect(Collectors.toList());

        placeRepository.saveAll(placesToSave);
        placeRepository.deleteAll(placesToDelete);

        log.info("[Place] 데이터 동기화 완료 - 추가/수정: {}, 삭제: {}", placesToSave.size(), placesToDelete.size());
    }

 

 

 

📝 어떻게 최적화할 것인가?

일단 왜 오래 걸리는 가에 대해 알아봅시다. 

현재 코드에서는 JPA의 saveAll()과 deleteAll()을 사용하여 데이터를 갱신하고 있습니다. 하지만 이 방식에는 몇 가지 성능적인 문제가 있습니다.

 

📌 saveAll()과 deleteAll()

Spring JPA는 saveAll()과 deleteAll()을 제공하여 대량의 데이터를 한 번에 추가하거나 삭제할 수 있도록 지원합니다.
보통 대량의 데이터 처리 시 개별 save()보다는 saveAll()을 사용하는 것이 일반적입니다.

그 이유는 다음과 같습니다.

✅ 개별 save()는 각각의 트랜잭션을 생성하여 실행되므로 성능이 저하될 수 있음
✅ 반면, saveAll()과 deleteAll()은 하나의 트랜잭션 내에서 실행되어 성능이 향상됨

 

하지만, JPA를 통한 대량 처리에는 한계가 있습니다.

  • JPA는 내부적으로 1건씩 INSERT/DELETE를 실행하기 때문에, SQL 배치 처리보다 성능이 낮음
  • 대량 삭제 시에도 DELETE FROM table WHERE id = ?를 반복 실행하여 성능이 저하됨
  • 트랜잭션 메모리 관리가 비효율적일 가능성이 있음
@Transactional
public <S extends T> List<S> saveAll(Iterable<S> entities) {
    Assert.notNull(entities, "Entities must not be null");
    List<S> result = new ArrayList();
    Iterator var4 = entities.iterator();

    while(var4.hasNext()) {
        S entity = (Object)var4.next();
        result.add(this.save(entity)); // saveAll도 내부에서 save를 단건 실행함
    }

    return result;
}

JPA에서 이러한 문제가 발생하는 이유는 JPA의 영속성 때문입니다.

  • JPA는 영속성 컨텍스트를 통해 변경 감지를 수행
    • saveAll()을 실행하면, JPA는 엔터티를 영속성 컨텍스트에 등록하고 변경 사항을 추적
    • 이후 트랜잭션이 끝날 때 flush()를 실행하여 DB에 반영
  • 대량의 데이터를 처리할 때 영속성 컨텍스트가 비효율적
    • 엔터티가 많아질수록 메모리 사용량 증가 → OutOfMemory(메모리 부족) 가능성
    • 쓰기 성능이 떨어짐 → 한 번에 많은 엔터티를 관리해야 하기 때문
    • deleteAll()도 N번의 DELETE 쿼리를 실행 → 단일 DELETE FROM보다 비효율적

 

즉 JPA는 영속성 컨텍스트(Persistence Context)에 데이터를 유지하려고 해서 대량 처리 시 메모리 사용량이 증가하고, 불필요한 DB I/O가 발생하는 것입니다.

📌 기존 코드의 DB I/O 계산해보기

기존 코드의 DB I/O를 계산해보면 무엇이 문제인지 알 수 있습니다.

더보기 
 @Override
    public void run(ApplicationArguments args) throws Exception {
        if (placeRepository.count() > 0) {
            log.info("[Place] 기존 데이터 갱신 시작");
        } else {
            log.info("[Place] 더미 데이터 삽입 시작");
        }
        importPlace();
    }

    private void importPlace() {
        Map<String, Place> csvDataMap = new HashMap<>();
        syncPlaceData("data/병원정보.csv", 1, 28, 29, 3, 10, 11, csvDataMap);
        syncPlaceData("data/약국정보.csv", 1, 13, 14, 3, 10, 11, csvDataMap);
        updateDatabase(csvDataMap);
    }

    private void syncPlaceData(String filePath, int nameIdx, int longitudeIdx, int latitudeIdx, int placeTypeIdx,
                               int addressIdx, int telIdx, Map<String, Place> csvDataMap) {
        try (
             InputStream inputStream = getClass().getClassLoader().getResourceAsStream(filePath);
             InputStreamReader reader = new InputStreamReader(inputStream, StandardCharsets.UTF_8);
             CSVReader csvReader = new CSVReader(reader)) {

            csvReader.readNext(); // 첫 번째 줄 헤더 건너뜀

            String[] nextLine;
            while ((nextLine = csvReader.readNext()) != null) {
                String name = nextLine[nameIdx];
                String placeTypeStr = nextLine[placeTypeIdx];
                String address = nextLine[addressIdx];
                String tel = nextLine[telIdx];

                Double longitude = null;
                Double latitude = null;

                try {
                    longitude = Double.parseDouble(nextLine[longitudeIdx]);
                    latitude = Double.parseDouble(nextLine[latitudeIdx]);
                } catch (NumberFormatException e) {
                    continue; // 좌표가 없으면 건너뜀
                }

                String uniqueKey = generateUniqueKey(name, address); // 유니크 키 생성
                if (csvDataMap.containsKey(uniqueKey)) {
                    log.warn("중복 데이터 발견 - 유니크 키: {}", uniqueKey);
                    continue;
                }
                Place place = Place.builder()
                    .name(name)
                    .place_type(Place_type.valueOf(placeTypeStr))
                    .address(address)
                    .tel(tel)
                    .coordinate(createPoint(latitude, longitude))
                    .build();

                csvDataMap.put(uniqueKey, place);
            }

        } catch (Exception e) {
            log.error("CSV 파일을 처리하는 중 오류 발생", e);
            throw new RuntimeException("CSV 파일을 처리하는 중 오류 발생", e);
        }
    }

    private void updateDatabase(Map<String, Place> csvDataMap) {
        // 기존 데이터 조회
        List<Place> existingPlaces = placeRepository.findAll();

        // 기존 데이터 맵 생성
        Map<String, Place> existingPlacesMap = existingPlaces.stream()
            .collect(Collectors.toMap(
                place -> generateUniqueKey(place.getName(), place.getAddress()),
                place -> place
            ));

        // 추가/수정 대상 찾기
        List<Place> placesToSave = csvDataMap.entrySet().stream()
            .filter(entry -> !existingPlacesMap.containsKey(entry.getKey()) || isUpdated(entry.getKey(), entry.getValue(), existingPlacesMap))
            .map(Map.Entry::getValue)
            .collect(Collectors.toList());

        // 삭제 대상 찾기
        List<Place> placesToDelete = existingPlaces.stream()
            .filter(place -> !csvDataMap.containsKey(generateUniqueKey(place.getName(), place.getAddress())))
            .collect(Collectors.toList());

        placeRepository.saveAll(placesToSave);
        placeRepository.deleteAll(placesToDelete);

        log.info("[Place] 데이터 동기화 완료 - 추가/수정: {}, 삭제: {}", placesToSave.size(), placesToDelete.size());
    }

    private boolean isUpdated(String key, Place newPlace, Map<String, Place> existingPlacesMap) {
        Place existingPlace = existingPlacesMap.get(key);

        if (existingPlace == null) {
            return true; // 기존 데이터가 없는 경우 업데이트 필요
        }

        // 필드별 비교
        return !Objects.equals(existingPlace.getName(), newPlace.getName())
            || !Objects.equals(existingPlace.getPlace_type(), newPlace.getPlace_type())
            || !Objects.equals(existingPlace.getAddress(), newPlace.getAddress())
            || !Objects.equals(existingPlace.getTel(), newPlace.getTel())
            || !existingPlace.getCoordinate().equalsExact(newPlace.getCoordinate());
    }

    private String generateUniqueKey(String name, String address) {
        return (name.trim() + "_" + address.trim()).replaceAll("\\s+", "_");
    }

    private Point createPoint(double latitude, double longitude) {
        Point point = geometryFactory.createPoint(new Coordinate(longitude, latitude));
        point.setSRID(4326);
        return point;
    }
}

 

 

  • JPA 영속성 컨텍스트를 사용하는 부분
    1. DB 저장 (saveAll()) → O(X) = 신규 및 변경 데이터(약 20,000개라 가정)
      • saveAll()이 2만 건의 INSERT 또는 UPDATE 실행
      • I/O 비용: 2만 회 (개별 SQL 실행 시)
    2. DB 삭제 (deleteAll()) → O(Y) = 삭제 데이터(약 1,000개라 가정)
      • deleteAll()이 1천 건의 DELETE 실행
      • I/O 비용: 1천 회 (개별 SQL 실행 시)

📌총 예상 I/O 비용 (JPA 기본 방식)

  • INSERT/UPDATE: 2만 회
  • DELETE: 1천 회

👉 총 약 21,000회 I/O 발생 ⚠️

 

📌개선 예상 (벌크 & Batch 적용 시)

  • INSERT/UPDATE: Batch(1,000개) → 20회 실행
  • DELETE: Batch(1,000개) → 1회 실행

👉 총 21회 I/O로 최적화 ✅

 

 

 

 

개선 방법은 두 가지로 생각해볼 수 있습니다.

  1. JDBC Batch 활용
  2. Spring Batch 도입

하지만 Spring Batch보다는 JDBC Batch가 더 적합하다고 판단했습니다.

우선, Spring Batch는 Spring의 외부 라이브러리입니다. 단순한 배치 작업을 위해 추가적인 의존성을 도입하고 Config를 설정하는 것보다 이미 제공되는 기능을 활용하는 것이 더 효율적이라고 생각했습니다.

또한, 처리해야 할 데이터 규모가 크지 않다는 점도 중요한 고려 요소입니다. 예상되는 데이터 건수는 10만 건 내외이며, 이 수치는 크게 증가할 가능성이 낮습니다. 이런 상황에서 Spring Batch를 도입하는 것은 오버 엔지니어링에 가깝다고 판단했습니다.

따라서 JDBC Batch를 활용하는 것이 가장 현실적인 선택이라 생각됩니다. 

 

📝JDBC Batch bulk Insert

 

왜 JPA 영속성을 제거한 bulk insert를 하지 않고 JDBC Batch bulk 를 사용하는가?

그 이유는 Place 엔티티의 place_id가 IDENTITY 전략을 사용하기 때문입니다.

IDENTITY 방식은 JPA에서 INSERT 시, 즉시 데이터베이스에서 ID 값을 받아오기 때문에, 쓰기 지연을 활용한 일괄 갱신이 불가능합니다.

따라서 JPA의 영속성을 제거한 벌크 연산을 수행하려면, Place 엔티티에서 id 필드를 제거하는 구조 변경이 필요합니다.

이는 부담이 크기 때문에, JDBC Batch bulk를 활용하여 쓰기 지연 없이 빠르게 갱신하도록 하겠습니다.

 

해당 JDBC Batchbulk는 JPA와 다르게 동작합니다. 

JPA 쿼리
insert into place (...) values (...)
insert into place (...) values (...)
insert into place (...) values (...)
insert into place (...) values (...)
JDBC Batch bulk
insert into place (...) 
	values 
	(...),
	(...),
	(...),
	(...)

 

해당 쿼리로 변경되며 한번에 여러개의 INSERT를 진행해 네트워크 및 DB 커넥션 자원이 감소됩니다.

 

그럼 한번 개선해보겠습니다.

    @Transactional
    protected void updateDatabase(Map<String, Place> csvDataMap) {
        long start = System.currentTimeMillis();
        // 기존 데이터 조회
        List<Place> existingPlaces = placeRepository.findAll();

        // 기존 데이터 맵 생성
        Map<String, Place> existingPlacesMap = existingPlaces.stream()
            .collect(Collectors.toMap(
                place -> generateUniqueKey(place.getName(), place.getAddress()),
                place -> place
            ));

        // 추가/수정 대상 찾기
        List<Place> placesToSave = csvDataMap.entrySet().stream()
            .filter(entry -> !existingPlacesMap.containsKey(entry.getKey()) || isUpdated(entry.getKey(), entry.getValue(), existingPlacesMap))
            .map(Map.Entry::getValue)
            .collect(Collectors.toList());

        // 삭제할 데이터 필터링
        List<Long> placeIdsToDelete = existingPlaces.stream()
            .filter(place -> !csvDataMap.containsKey(generateUniqueKey(place.getName(), place.getAddress())))
            .map(Place::getId)
            .collect(Collectors.toList());

        // ✅ 벌크 INSERT & DELETE 수행
        placeBulkRepository.batchInsertPlaces(placesToSave);
        placeBulkRepository.batchDeletePlaces(placeIdsToDelete);


        long end = System.currentTimeMillis();
        log.info("[Place] 데이터 동기화 완료 - 추가/수정: {}, 삭제: {}, 시간: {}", placesToSave.size(), placeIdsToDelete.size(), end - start);
    }

변경된 점:

  1. saveAll() → batchInsertPlaces() (JDBC Batch Insert 활용)
  2. deleteAll() → batchDeletePlaces() (벌크 삭제)
  3. 메소드에 @Transactional을 적용해 -> 하나의 트랜잭션으로 묶어 리소스 최적화

 

자세히 내부 동작이 변경된 것을 예시로 들면 개별로 적용되던 쿼리가 다음처럼 적용됩니다.

 batchInsertPlaces() 
    public void batchInsertPlaces(List<Place> places) {
        if (places.isEmpty()) return;

        String sql = "INSERT INTO place (name, place_type, address, tel, coordinate) " +
            "VALUES (?, ?, ?, ?, ST_GeomFromText(?, 4326))";

        jdbcTemplate.batchUpdate(sql, places, BULK_COUNT, (ps, place) -> {
            ps.setString(1, place.getName());
            ps.setString(2, place.getPlace_type().name());
            ps.setString(3, place.getAddress());
            ps.setString(4, place.getTel());
            ps.setString(5, String.format("POINT(%f %f)", place.getCoordinate().getY(), place.getCoordinate().getX()));
        });

        log.info("[PlaceBulkRepository] {} 개의 Place 데이터 벌크 삽입 완료", places.size());
    }
INSERT INTO place (name, place_type, address, tel, coordinate) 
VALUES 
    ('카페A', 'CAFE', '서울 강남구', '010-1234-5678', ST_GeomFromText('POINT(127.123 37.456)', 4326)),
    ('카페B', 'CAFE', '서울 마포구', '010-5678-1234', ST_GeomFromText('POINT(126.987 37.543)', 4326)),
    ('카페C', 'CAFE', '서울 종로구', '010-1111-2222', ST_GeomFromText('POINT(127.000 37.000)', 4326));

 

batchDeletePlaces()
  // ✅ 벌크 DELETE
    public void batchDeletePlaces(List<Long> placeIds) {
        if (placeIds.isEmpty()) return;

        String sql = "DELETE FROM place WHERE id = ?";

        jdbcTemplate.batchUpdate(sql, placeIds, BULK_COUNT, (ps, id) -> ps.setLong(1, id));

        log.info("[PlaceBulkRepository] {} 개의 Place 데이터 벌크 삭제 완료", placeIds.size());
    }
DELETE FROM place WHERE id IN (1001, 1002, 1003, ...);

 

 

📌결과

  • 기존
[Place] 데이터 동기화 완료 - 추가/수정: 100676, 삭제: 0, 시간: 172223ms
[Place] 데이터 동기화 완료 - 추가/수정: 100676, 삭제: 0, 시간: 68970ms
  • 개선 후
[Place] 데이터 동기화 완료 - 추가/수정: 100676, 삭제: 0, 시간: 139602ms
[Place] 데이터 동기화 완료 - 추가/수정: 100676, 삭제: 0, 시간: 37439ms
  JPA JDBC Batch Bulk 성능
인덱스 o 172초 139초 약 19.2% 성능 향상
인덱스 x 68초 37초 약 45.59% 성능 향상

 

유의미하게 성능이 향상 되는 것을 알 수 있습니다. 

 

 

🧷결론

 

기존에 saveAll() 방식을 사용하던 상황에서, JDBC Batch Bulk 연산을 도입하여 19.2%에서 최대 45.59%까지 성능 향상을 이루어냈습니다.

이 결과가 나온 이유는, JPA를 맹신하지 않고 각 기술의 한계를 객관적으로 검토했기 때문입니다. 세상에는 완벽한 기술은 없으며, 모든 기술은 저마다의 장단점을 가지고 있습니다. 이번 포스팅에서는 특히 JPA의 영속성 문제로 인해 대량 데이터 처리 시 발생하는 성능 저하가 두드러졌음을 확인할 수 있었습니다.

실버 불릿:
"소프트웨어 개발에 있어 단 하나의 '실버 불릿'은 존재하지 않습니다. 상황에 따라 올바른 도구와 접근법을 선택하는 것이 최선의 해결책입니다."

이러한 관점을 갖게 된 계기는(  Hibernate 의 ‘불편한’ 편의 기능들. Hibernate 는 지연 로딩, 캐싱, 쿼리 순서 최적화 등 개발자가… | by Hyeon9mak | Monday9pm )라는 글을 읽으면서입니다.

좋은 개발자로 성장하기 위해서는, 이러한 생각을 당연하게 받아들이고 상황에 맞게 기술을 선택하는 능력이 필요한 것 같습니다. 

감사합니다.

'Project > PlusKM' 카테고리의 다른 글

[+/KM]공간 데이터 데이터베이스 최적화를 위한 MongoDB 도입  (0) 2025.01.22
[+/KM]CSV 파일 DB에 유지하기  (0) 2025.01.21
[+/KM]MySQL에서 공간 데이터 최적화하기  (0) 2025.01.20
'Project/PlusKM' Related Articles more