[+/KM]AWS OutOfMemoryError 극복: active 컬럼 기반 공공데이터 갱신 배치 전략

📝배경

저번 포스팅 에서 JDBC Batch bulk Insert을 이용해서 공공데이터 갱신 작업 최적화를 진행하였습니다.

갱신 작업 자체는 로컬 호스트에서 성공적으로 성능이  개선되었으나... AWS에서 배포 시에 문제가 발생했습니다. 

 

해당 오류는 이 오류 메시지는 java.lang.OutOfMemoryError: Java heap space 즉, Java 힙 메모리가 부족해서 발생한 오류입니다. 자세히 살펴보면, Tomcat 서버의 NIO 커넥션 폴링 스레드에서 ConcurrentHashMap의 iterator를 돌리다가 메모리가 부족해진 상황입니다.

@Slf4j
@RequiredArgsConstructor
@Order(1)
@DummyDataInit
public class PlaceInitializer implements ApplicationRunner {
    private final PlaceRepository placeRepository;
    private final GeometryFactory geometryFactory;
    private final PlaceBulkRepository placeBulkRepository;
    @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);
        }
    }
    @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,
                (existing, duplicate) -> duplicate // 중복된 키가 있을 경우 최신 데이터를 유지
            ));

        // 추가/수정 대상 찾기
        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);

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

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

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

    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;
    }
}

 

위의 코드는 문제가 터진 코드의 전체입니다. 

 

어디서 문제가 발생하는지 생각해봅시다. 

생각할 수 있는 주요 원인은 하나의 Map에 모든 공공데이터를 넣고 작업을 진행하기 때문입니다. 

 

1. placeRepository.findAll() 호출
   • DB에 저장된 모든 Place 데이터를 한꺼번에 전부 조회해서 리스트에 올립니다.
   • 데이터가 많으면 JVM 힙에 큰 부담이 될 수 있습니다.
2. csvDataMap에 모든 CSV 데이터를 한꺼번에 저장
   • CSV 파일에서 읽은 데이터를 한꺼번에 메모리에 다 올립니다.
   • 특히 CSV 데이터가 수십만 건 이상이라면 메모리 사용량이 급격히 올라갑니다.
3. 비교 작업도 모두 메모리 내 Map을 이용
   • 기존 DB 데이터와 CSV 데이터를 Map으로 변환해 비교하는 과정 역시 메모리 점유가 큽니다.
4. 벌크 삽입/삭제를 한 번에 처리
   • placeBulkRepository.batchInsertPlaces(placesToSave) 같은 벌크 작업도 메모리에 큰 데이터를 쌓고 작업할 수 있습니다.

---

 

즉 하나의 Map에 수십만건의 데이터를 넣고 진행하기에 메모리 사용량이 급격히 올라간 것입니다.

 

 

 

📝 해결 방법 1 : JVM 힙 메모리 증설

 

현재 저가 사용하는 AWS 서버는 프리티어 t3.micro 입니다. 

t3.micro 스팩을 한번 살펴보면

vCPU	메모리	CPU 성능	CPU 크래딧	네트워크 성능
2개	1GB	기본 성능의 10%	시간당 12	최대 5 Gbps

 

 

최대 1GB의 메모리를 가지고 있습니다. 그렇기에 JVM 힙 메모리를 최대한 늘려보겠습니다.

 

• 전체 메모리 1GB 중 OS 및 기타 프로세스에 최소 200~300MB는 남겨두는 게 안전하다고 생각하였습니다.
• JVM 힙 메모리는 약 700~750MB 정도로 설정하는 게 무난하다고 판단했습니다.

docker-compose.yml 파일

    environment:
      - TZ=Asia/Seoul
      - JAVA_OPTS=-Xms512m -Xmx700m

 

 

해당 코드처럼 증설을 해줬습니다. 

 

그러나 다시 실행해보니 똑같은 OOM 오류로 실패하였습니다.

 

📝 해결 방법 2 : Batch 전략 

현재 사용 중인 t3.micro 인스턴스는 프리 티어에 해당되며, 상위 스펙으로 업그레이드할 경우 비용적인 부담이 있어 현실적으로 어려운 상황입니다. 따라서 제한된 메모리 환경에서도 안정적으로 동작할 수 있도록 JVM 메모리 사용을 최소화하고 애플리케이션을 경량화하는 방향으로 접근하고자 합니다

 

 

✅ 왜 배치 전략이 메모리 사용을 줄일 수 있을까?

현재 코드는 다음과 같은 방식입니다:

• CSV 전체 데이터를 Map<String, Place>에 다 넣음 → 메모리 사용량 증가
• 기존 DB 데이터를 List<Place> → Map<String, Place>에 전부 로딩 → 메모리 사용량 증가
• 그리고 placesToSave, placeIdsToDelete 같은 중간 컬렉션도 유지됨 → 힙 점유도 증가

👉 즉, 모든 데이터를 한 번에 메모리에 올리는 방식은 작은 인스턴스(t3.micro 1GB RAM)에서 OOM(OutOfMemoryError)을 유발할 수 있음

 

 

이러한 메모리 문제를 근본적으로 해결하기 위해 배치 처리 전략으로의 전환이 필수적이라고 판단했습니다. 그러나 단순한 배치 전환으로는 기존의 효율적인 비교 로직을 유지하기 어렵습니다. 1,000개 단위로 데이터를 처리할 때는 전체 데이터를 한 번에 보고 비교하는 방식이 불가능하기 때문입니다

.

✅  active 컬럼 도입

이에 대한 해결책으로 place 테이블에 active라는 새로운 컬럼을 추가했습니다. 이는 데이터 동기화 과정에서 다음과 같은 방식으로 메모리 효율성을 극대화합니다:

1. 초기화 단계: 공공데이터 갱신 작업 시작 시, place 테이블 내 모든 기존 레코드의 active 값을 false로 일괄 업데이트합니다. 이 작업은 단일 쿼리로 이루어져 메모리 부담이 적습니다.
2. 배치 처리 및 갱신: CSV 파일에서 1,000개 단위로 데이터를 읽어오면서, 해당 데이터를 데이터베이스에 삽입(INSERT)하거나 기존 데이터를 갱신(UPDATE)합니다. 이때, CSV 파일에 존재하는 데이터는 active 값을 true로 설정합니다. ON DUPLICATE KEY UPDATE와 같은 기능을 활용하여 기존 레코드가 있다면 active를 true로 바꾸고, 새로운 레코드라면 active가 true인 상태로 삽입합니다.
3. 오래된 데이터 정리: 모든 CSV 데이터 처리가 완료된 후, 여전히 active 값이 false로 남아있는 레코드들을 일괄 삭제합니다. 이들은 이번 갱신 주기에서 CSV 파일에 존재하지 않아 활성화되지 않은, 즉 더 이상 유효하지 않은 오래된 데이터로 간주됩니다.

✅  형식 변경에 대한 부담 최소화

데이터베이스 테이블 형식을 변경하는 것은 일반적으로 신중을 기해야 하는 중요한 작업입니다. 그러나 이번 active 컬럼 도입은 그 부담을 최소화하면서 OutOfMemoryError 문제를 해결하는 최적의 방안입니다.

이 컬럼은 다음과 같은 이유로 기존 시스템에 미치는 영향을 극히 제한합니다:

• 용도 한정: active 컬럼은 오직 공공데이터 갱신 작업에만 사용됩니다. 애플리케이션의 핵심 비즈니스 로직이나 기존 데이터 모델과의 연관 관계에는 전혀 관여하지 않습니다.
• 기존 기능 무영향: 공간 조회와 같은 기존 기능들은 active 컬럼의 유무나 값에 영향을 받지 않고 독립적으로 작동합니다. 이는 현재 서비스 운영에 지장을 주지 않으면서 안정적인 전환을 가능하게 합니다.
• DDL 안정성: DDL(Data Definition Language) 과정에서도 예상되는 큰 오류가 없어, 실제 배포 시 발생할 수 있는 위험을 효과적으로 줄일 수 있습니다.

 

바뀐 공공데이터 갱신 코드

@Slf4j
@RequiredArgsConstructor
@Order(1)
@DummyDataInit
public class PlaceInitializer implements ApplicationRunner {

    private final PlaceRepository placeRepository;
    private final GeometryFactory geometryFactory;
    private final PlaceBulkRepository placeBulkRepository;

    private static final int BATCH_SIZE = 1000;

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

        // 1. 기존 모든 Place 데이터를 비활성화합니다.
        placeRepository.deactivateAll();

        // 2. CSV 데이터를 스트리밍 방식으로 읽어와 DB에 삽입/업데이트합니다.
        importPlaceStreaming();

        // 3. `importPlaceStreaming()`을 통해 갱신되지 않아 여전히 `active=false` 상태인
        //    모든 Place 데이터를 삭제합니다.
        placeRepository.deleteInactivePlaces();

        log.info("[Place] 더미 데이터 삽입/갱신 완료");
    }

    private void importPlaceStreaming() {
        Map<String, Place> csvDataMap = new HashMap<>();

        // 병원 및 약국 CSV 파일 데이터를 동기화합니다.
        syncPlaceData("data/병원정보.csv", 1, 28, 29, 3, 10, 11, csvDataMap);
        syncPlaceData("data/약국정보.csv", 1, 13, 14, 3, 10, 11, csvDataMap);

        if (!csvDataMap.isEmpty()) {
            updateDatabasePartial(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;
                Double latitude;

                try {
                    longitude = Double.parseDouble(nextLine[longitudeIdx]);
                    latitude = Double.parseDouble(nextLine[latitudeIdx]);
                } catch (NumberFormatException e) {
                    continue;
                }

                String uniqueKey = generateUniqueKey(name, address);
                if (csvDataMap.containsKey(uniqueKey)) {
                    continue;
                }

                if (placeTypeStr != null && !placeTypeStr.trim().isEmpty()) {
                    Place place = Place.builder()
                        .name(name)
                        .place_type(Place_type.valueOf(placeTypeStr))
                        .address(address)
                        .tel(tel)
                        .active(true)
                        .coordinate(createPoint(latitude, longitude))
                        .build();

                    csvDataMap.put(uniqueKey, place);
                } else {
                    continue;
                }


                // BATCH_SIZE만큼 데이터가 쌓이면 부분적으로 DB에 반영하고 맵을 비웁니다.
                if (csvDataMap.size() >= BATCH_SIZE) {
                    updateDatabasePartial(csvDataMap);
                    csvDataMap.clear();
                }
            }

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

    @Transactional
    protected void updateDatabasePartial(Map<String, Place> csvDataMap) {
        if (csvDataMap.isEmpty()) {
            return;
        }

        List<String> keys = new ArrayList<>(csvDataMap.keySet());
        // 고유 키(이름 + 주소 조합)로 기존 Place 데이터를 조회합니다.
        List<Place> existingPlaces = placeRepository.findAllByUniqueKeyIn(keys);

        Map<String, Place> existingMap = existingPlaces.stream()
            .collect(Collectors.toMap(
                p -> generateUniqueKey(p.getName(), p.getAddress()),
                p -> p
            ));

        // 삽입 또는 업데이트가 필요한 데이터를 선별합니다.
        List<Place> toSave = csvDataMap.entrySet().stream()
            .filter(entry -> !existingMap.containsKey(entry.getKey()) || isUpdated(entry.getKey(), entry.getValue(), existingMap))
            .map(Map.Entry::getValue)
            .collect(Collectors.toList());

        if (!toSave.isEmpty()) {
            placeBulkRepository.batchInsertPlaces(toSave);
        }
        log.info("[Place] 배치 업데이트 - 처리건수: {}", toSave.size());
    }

    // 기존 Place와 비교하여 업데이트가 필요한지 확인합니다.
    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+", "_");
    }

    // 위도, 경도 정보를 사용하여 Point 객체를 생성하고 SRID를 설정합니다.
    private Point createPoint(double latitude, double longitude) {
        Point point = geometryFactory.createPoint(new Coordinate(longitude, latitude));
        point.setSRID(4326);
        return point;
    }
}

 

 

 

 

JDBC Batch Bulk 메소드 변경

    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());
    }

    public void batchInsertPlaces(List<Place> places) {
        if (places.isEmpty()) return;
        String sql = "INSERT INTO place (name, place_type, address, tel, active, coordinate) " +
            "VALUES (?, ?, ?, ?, ?, ST_GeomFromText(?, 4326)) " +
            "ON DUPLICATE KEY UPDATE " +
            "place_type = VALUES(place_type), " +
            "address = VALUES(address), " +
            "tel = VALUES(tel), " +
            "active = VALUES(active), " +
            "coordinate = VALUES(coordinate)";

        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.setBoolean(5, place.isActive());
            ps.setString(6, String.format("POINT(%f %f)", place.getCoordinate().getX(), place.getCoordinate().getY()));
        });


        log.info("[PlaceBulkRepository] {} 개의 Place 데이터 벌크 삽입 완료", places.size());
    }

 

ON DUPLICATE KEY UPDATE를 사용하게 되면 Bulk Delete 메소드가 사라지게 되었습니다. 

추가한 active 컬럼에 의해 필요가 없게 되었기 때문입니다.

 

 

🧷결론 

이번 포스팅은 AWS t3.micro 인스턴스에서 발생했던 OutOfMemoryError를 해결하고, 대용량 공공데이터 갱신 작업의 안정성과 효율성을 확보하기 위한 여정을 담았습니다.

 

초기에 JDBC Batch Bulk Insert를 통한 성능 개선은 로컬 환경에서 성공적이었으나, 1GB 메모리라는 t3.micro의 제약 조건 속에서는 모든 데이터를 메모리에 올려 비교하고 처리하는 방식이 결국 OutOfMemoryError를 유발했습니다. JVM 힙 메모리를 최대로 증설하는 시도조차 근본적인 해결책이 될 수 없음을 확인했습니다.

이에 우리는 메모리 사용량을 최소화하는 배치 처리 전략으로의 전환을 결정했습니다. 특히, 이 전략의 핵심은 place 테이블에 active라는 새로운 컬럼을 도입하는 것이었습니다.

 

이러한 active 컬럼 기반의 배치 전략은 기존의 Bulk Delete 메서드를 제거하며 코드의 복잡성을 줄였고, 모든 데이터를 메모리에 로드하지 않아 OutOfMemoryError를 성공적으로 해결했습니다. 또한, 이 active 컬럼은 오직 공공데이터 갱신 작업에만 사용되고 기존 서비스의 연관 관계나 공간 조회 로직에는 영향을 미치지 않으므로, 데이터베이스 테이블 형식 변경에 대한 부담도 최소화했습니다.

 

 

quddaz/plusKM-BE: 내 주변 병원과 약국 위치 파악과 가격 비교를 통한 현명한 소비 +/Km