[토이프로젝트] KBO 경기 일정 수집하기(2) - Upsert를 통한 중복 데이터 처리

목차

/pages/KBO-경기-일정-수집하기-시리즈

Upsert의 필요성

1편에서 경기 정보를 수집하고 DB에 저장했지만 경기 정보는 매일매일 바뀌게됩니다.

 

시즌 초에 전체 일정이 나오지만 매일 경기를 진행하면서 경기 정보는 수시로 바뀝니다. 게다가 날씨에 따라 취소 경기가 발생하여 더블헤더 경기가 편성되거나 추후 잔여경기 일정으로 편성될 수도 있죠.

 

따라서 수집된 경기 목록 중에서 DB에 저장된 경기는 업데이트하고 새로운 경기만 저장하는 Upsert 전략이 필요합니다.

Upsert 전략을 구성하자!

우선 어떻게 Upsert 전략을 구성할지 생각해봤습니다.

KBO 웹사이트에서 경기 정보를 수집하여 파싱하면 경기 정보 엔티티 객체가 모인 리스트가 생성됩니다. 이 경기 정보 객체들과 중복되는 레코드가 DB에 저장되어 있는지 확인하고, 조회된 중복 레코드에 한해 새로 수집된 경기 정보로 업데이트 할 수 있겠네요.

 

한가지 더 고려해보자면 현재 경기 정보를 수집하는 기능은 REST API를 수동으로 호출하도록 되어있습니다. 경기 정보 수집 API를 호출하면 KBO 웹사이트에서 경기 정보를 수집하고, DB에 저장하는 방식이죠.

 

아무래도 수동으로 API를 호출할 수 있게 되어있으니 확실한 응답을 주는것도 중요해 보였습니다. 그래서 경기 정보를 수집할 때 아래와 같이 수집된 데이터의 개수, 저장된 레코드의 개수, 수정된 레코드의 개수를 보여줄 수 있으면 좋을 것 같았습니다.

{
    "collectedCount": 144,
    "savedCount": 4,
    "modifiedCount": 140
}

다만 이 경우에는 수정된 레코드의 개수를 보여주려면 실제로 경기 정보가 수정되었는지를 체크할 수 있어야 합니다. DB에 저장된 경기 정보와 새로 수집한 경기 정보가 완전히 일치하는 경우를 걸러내야 하는 것이죠.

 

정리하자면 수집된 경기 정보와 중복되는 레코드가 있는지 조회하고, 중복되는 레코드는 변경점이 있는지도 체크할 수 있어야 합니다.

어떻게 구현할까?

저장된 경기정보를 조회하자

1편에서 사용한 경기 정보 엔티티의 구조를 다시 보겠습니다.

class GameInfo(
    val gameId: Long? = null,       // DB 식별자(PK)
    val gameKey: String,        // 실제로 경기 정보를 구분할 UK
    val seriesType: SeriesType,     // 시범경기, 정규시즌, 포스트시즌
    val date: LocalDate,
    var time: LocalTime?,   // 더블헤더 경기는 시간이 미정이므로 nullable
    val awayTeam: Team,
    val homeTeam: Team,
    
    // 경기가 시작하지 않았거나 취소되면 점수가 없으므로 null
    var awayScore: Int? = null,
    var homeScore: Int? = null,

    var relay: String?,
    var stadium: String,
    var gameStatus: GameStatus,     // 예정됨, 진행중, 종료, 취소
    var cancellationReason: CancellationReason?     // gameStatus가 CANCELLED일 때만 존재
)

여기서 gameId는 DB 상에서 데이터를 구분하기 위한 식별자이고, 실제로 각 경기를 구분할 때는 gameKey를 사용했습니다.

 

gameKey는 {yyyyMMdd}-{원정팀}-{홈팀}-{일련번호}의 형태가 되는 만큼, 이를 통해 DB에 중복 데이터가 존재하는지 체크할 수 있습니다.

 

경기 정보를 수집하여 파싱하면 경기 정보의 리스트가 됩니다. 이를 통해 수집한 경기 정보들의 gameKey 리스트를 추출하여 DB에 이미 저장된 경기들의 엔티티 객체를 손쉽게 조회할 수 있습니다.

val existingGamesMap = gameInfoRepository.findByGameKeyIn(gameInfoList.map { it.gameKey })
    .associateBy { it.gameKey }

findByGameKeyIn()을 통해 gameInfoList에 담긴 경기 정보들의 gameKey와 일치하는 경기들의 목록을 DB에서 조회합니다. 그리고 이미 존재하는 경기 정보 리스트에 associateBy를 호출하여 gameKey로 경기 정보를 조회할 수 있는 맵을 생성합니다.

 

이렇게 생성된 맵에서 수집된 경기의 gameKey로 조회되는 결과가 있다면 해당 경기는 중복되는 레코드라고 볼 수 있습니다.

중복 경기의 정보를 갱신하자

수집한 경기 정보와 중복되는 레코드를 모두 조회했으니 이젠 이 중복 레코드와 수집한 경기 정보 사이에 변경점이 발생했는지 체크해야 합니다.

 

우선 그 전에 DB상의 데이터 변경은 모두 JPA의 변경 감지를 통해 이뤄지는 만큼, 먼저 GameInfo 클래스에 엔티티의 프로퍼티를 수정하는 update() 메소드를 만들었습니다.

fun update(newGameInfo: GameInfo) {
    this.time = newGameInfo.time
    this.awayScore = newGameInfo.awayScore
    this.homeScore = newGameInfo.homeScore
    this.stadium = newGameInfo.stadium
    this.relay = newGameInfo.relay
    this.gameStatus = newGameInfo.gameStatus
    this.cancellationReason = newGameInfo.cancellationReason
}

앞서 수집한 경기 정보와 중복되는 레코드들을 조회했으니 이를 기반으로 새로운 경기와 업데이트할 경기를 분류합니다. 이후 새로운 경기 정보는 모두 저장하고 업데이트할 경기는 경기 정보를 순회하면서 update()를 통해 새로운 정보로 갱신해주면 됩니다.

// existingGamesMap를 기반으로 새로운 경기와 업데이트할 경기 분류
val (newGameList, gameListToUpdate) = gameInfoList.partition { existingGamesMap[it.gameKey] == null }
// 새로운 경기는 모두 저장
gameInfoRepository.saveAll(newGameList)
// 업데이트할 경기는 모두 업데이트
for (gameInfo in gameListToUpdate) {
    existingGamesMap[gameInfo.gameKey]?.update(gameInfo)
}

실제로 변경점이 발생했는지를 체크하자

여기까지만 해도 경기 정보 수집이라는 목적은 달성했습니다. 하지만 경기 정보 수집 API의 응답을 구성하려면 JPA의 변경 감지가 작동하여 실제로 UPDATE 쿼리가 발생했는지를 체크해야 합니다.

 

이건 간단하게 GameInfo의 update 로직을 조금 수정하는 것으로 구현했습니다.

fun update(newGameInfo: GameInfo): Boolean {
    val isUpdated = this.time != newGameInfo.time
            || this.awayScore != newGameInfo.awayScore
            || this.homeScore != newGameInfo.homeScore
            || this.stadium != newGameInfo.stadium
            || this.relay != newGameInfo.relay
            || this.gameStatus != newGameInfo.gameStatus
            || this.cancellationReason != newGameInfo.cancellationReason

    if (!isUpdated) {
        return false
    }

    this.time = newGameInfo.time
    this.awayScore = newGameInfo.awayScore
    this.homeScore = newGameInfo.homeScore
    this.stadium = newGameInfo.stadium
    this.relay = newGameInfo.relay
    this.gameStatus = newGameInfo.gameStatus
    this.cancellationReason = newGameInfo.cancellationReason

    return true
}

먼저 엔티티 객체의 변경이 있는지 확인하기 위한 isUpdated라는 변수를 선언합니다. 이 변수는 var로 선언된 모든 프로퍼티들을 비교하여 그 중 하나라도 원본과 다르다면 true가, 그렇지 않다면 false가 됩니다. 그리고 이 변수를 기반으로 업데이트를 진행합니다.

 

앞서 구현한 Upsert 로직도 조금 수정해줍니다.

for (gameInfo in gameListToUpdate) {
    existingGamesMap[gameInfo.gameKey]?.run {
        if (update(gameInfo)) modifiedCount++
    }
}

엔티티 객체의 update를 실행하고, 그 결과에 따라 실제로 변경된 경기 정보의 수를 나타내는 modifiedCount를 증가시키도록 했습니다.

동시성 이슈?

엄밀히 따지면 이 코드는 동시성 이슈가 발생할 수 있는 코드입니다. 예를 들면 다음과 같은 상황이 발생할 수 있죠.

 

1. 스레드A가 트랜잭션을 시작하고 existingGamesMap을 조회합니다. (DB값: awayScore=1)
2. 스레드B도 동시에 들어와서 existingGamesMap을 조회합니다. (DB값: awayScore=1)
3. 스레드B가 먼저 awayScore=2로 업데이트하고 커밋합니다. (DB값: awayScore=2로 변경됨)
4. 스레드A는 자신의 1차 캐시에 있는 엔티티(여전히 awayScore=1)를 보고 있습니다.
5. 스레드A는 가져온 데이터(awayScore=2)와 1차 캐시의 엔티티(awayScore=1)를 비교합니다.
   • update() 실행: 엔티티 프로퍼티 값을 2로 바꿈.
   • update() 실행 결과 : true
   • modifiedCount++가 실행됨
6. 스레드A가 커밋합니다.
   • JPA는 변경 감지를 통해 업데이트 쿼리를 날립니다.
   • DB에선 이미 awayScore=2였지만, 다시 2로 덮어씌워집니다.

 

JPA의 변경 감지는 1차 캐시에 영속화된 스냅샷과 비교하여 작동합니다. 따라서 위 시나리오대로면 스레드A의 1차 캐시에는 스레드B가 커밋한 결과가 반영되지 않죠. 이 결과 스레드A에서 실행한 경기 정보 수집 결과에서는 실제로 수정된 경기 정보가 없지만(스레드B에서 이미 같은 정보로 수정해버렸으니) modifiedCount가 1이 되죠.

 

하지만 경기 정보 수집이라는 기능적 특성 상, 스케줄러를 통해 주기적으로 실행되거나 제한된 횟수로만 수동으로 실행하기 때문에 동시성 이슈 발생 가능성이 사실상 없다고 판단했습니다.

 

또한 동시에 실행되더라도 결국 같은 정보를 수집하기 때문에 데이터 정합성에서도 문제가 없으니 동시성 이슈를 처리하는건 오버엔지니어링이라는 생각이 들어 이대로 구현했습니다.

Upsert 코드

GameInfoDataService.kt

@Transactional
fun saveOrUpdateGameInfo(gameInfoList: List<GameInfo>): CollectDataResponse {
    val existingGamesMap = gameInfoRepository.findByGameKeyIn(gameInfoList.map { it.gameKey })
        .associateBy { it.gameKey }

    val (newGameList, gameListToUpdate) = gameInfoList.partition { existingGamesMap[it.gameKey] == null }
    gameInfoRepository.saveAll(newGameList)

    var modifiedCount = 0
    for (gameInfo in gameListToUpdate) {
        existingGamesMap[gameInfo.gameKey]?.run {
            if (update(gameInfo)) modifiedCount++
        }
    }

    return CollectDataResponse(gameInfoList.size, newGameList.size, modifiedCount)
}

GameInfo.kt

fun update(newGameInfo: GameInfo): Boolean {
    val isUpdated = this.time != newGameInfo.time
            || this.awayScore != newGameInfo.awayScore
            || this.homeScore != newGameInfo.homeScore
            || this.stadium != newGameInfo.stadium
            || this.relay != newGameInfo.relay
            || this.gameStatus != newGameInfo.gameStatus
            || this.cancellationReason != newGameInfo.cancellationReason

    if (!isUpdated) {
        return false
    }

    this.time = newGameInfo.time
    this.awayScore = newGameInfo.awayScore
    this.homeScore = newGameInfo.homeScore
    this.stadium = newGameInfo.stadium
    this.relay = newGameInfo.relay
    this.gameStatus = newGameInfo.gameStatus
    this.cancellationReason = newGameInfo.cancellationReason

    return true
}

깃허브 링크

GitHub - Colabear754/kbo-scraper: KBO 경기 일정을 크롤링하는 프로젝트