Data Lab

멱등성과 트랜잭션: 중복 실행에도 안전한 데이터 처리 설계

1. 왜 멱등성과 트랜잭션을 공부하게 되었는가?Game Data Project #21에서는 Steam News, PUBG, Overwatch 패치노트 수집을 Airflow에 연결했다.그리고 정기 실행과 부분 실패 감지까지 구현했다.이 과정에서 한 가지 고민이 생겼다."데이터 수집이 중간에 실패해서 다시 실행되면 이미 저장한 데이터는 어떻게 될까?"예를 들어 다음 상황을 생각해 볼 수 있다.1. Steam News 수집 성공2. PostgreSQL에 신규 뉴스 저장 성공3. PUBG 수집 도중 Timeout 발생4. Airflow Task FAILED5. Airflow에서 Task 재실행6. Steam News를 다시 수집이때 이미 저장한 Steam News가 또 저장된다면 중복 데이터가 발생한다.또 다른..

Game Data Project #21-2 - Airflow 패치 수집 안정화와 정기 실행

#21-1에서는 기존 패치 수집 파이프라인을 Airflow에 연결했다.Steam News, PUBG, Overwatch의 패치노트를 수집하고, 새로운 패치를 감지한 뒤 LLM으로 번역·요약하는 기존 로직을 하나의 Airflow Task에서 실행하도록 구성했다.실제 airflow dags test 명령으로 수집이 정상적으로 실행되는 것까지 확인했다.하지만 아직 정기 실행을 활성화하지는 않았다.기존 Collector가 일부 오류를 내부에서 처리하는 구조였기 때문에, 실제 수집에 문제가 발생했는데도 Airflow에는 SUCCESS로 표시될 가능성이 있었기 때문이다.이번 #21-2에서는 이 문제를 해결하고, LLM 처리량과 Retry 정책을 정리한 뒤 실제 정기 실행까지 활성화했다.1. Airflow의 SUCC..

Game Data Project #21-1 - Airflow로 패치 수집 자동화하기

#20에서는 기존 Player Count와 Price Collector를 Airflow에 연결하고, 실패 상황에 따른 Skip과 Retry 정책까지 구성했다.이제 사용자가 게임을 등록하면 Airflow가 정해진 시간에 데이터를 수집하고, 일부 게임에서 오류가 발생하더라도 가능한 범위에서 수집을 계속할 수 있다.하지만 아직 자동화되지 않은 수집 파이프라인이 하나 남아 있었다.바로 패치노트 수집 파이프라인이다.패치노트는 이미 Steam News, PUBG, Overwatch를 대상으로 수집하고 있었고, 새로운 패치를 감지하면 LLM을 통해 한국어 번역과 요약까지 수행할 수 있었다.문제는 실행 방식이었다.기존에는 Docker Compose를 시작할 때 별도의 초기화 컨테이너를 통해 패치 수집을 실행하는 구조였..

#20-1에서는 기존 Player Count와 Price Collector를 Airflow와 연결했다.수집 주기는 다음과 같이 구성했다.Player Count→ 매시간Price→ 6시간마다그리고 모든 Collector가 games 테이블을 기준으로 수집 대상을 가져오기 때문에 Web에서 새로운 게임을 등록하면 별도의 코드 수정 없이 다음 Airflow 실행부터 자동으로 수집 대상에 포함된다.하지만 자동으로 실행된다고 해서 바로 안정적인 데이터 파이프라인이 되는 것은 아니다.실제 운영에서는 외부 API 요청이 항상 성공하지 않는다.API 404TimeoutHTTP 5xxNetwork ErrorDB Error잘못된 응답 데이터같은 상황이 계속 발생할 수 있다.이번 작업에서는 이런 오류들을 모두 똑같이 처리하..

백엔드 시스템을 만들다 보면 비슷해 보이는 용어들이 계속 등장한다. SchedulerWorkerQueueOrchestrator 처음 보면 모두: “작업을 자동으로 실행해주는 것” 처럼 느껴진다.하지만 실제 역할은 꽤 다르다.이번에 Game Data Project에서 Airflow로 Player Count와 Price Collector를 자동 실행하면서 이 차이를 이해하기 좋은 시점이 되었다.현재 프로젝트를 단순화하면: Airflow ↓Collector ↓Steam API ↓PostgreSQL 구조다.여기서 질문이 생긴다. Airflow는 Scheduler인가?Worker인가?Queue인가?Orchestrator인가?Collector는 Worker인가?나중에 Discord 알림을 만들 때..

#19에서는 사용자가 Steam Store URL을 입력하면 게임이 games 테이블에 등록되고, 기존 Collector가 해당 게임을 자동으로 인식하도록 만들었다.하지만 한 가지는 여전히 수동이었다.게임을 등록한 뒤 실제 데이터를 수집하려면 직접 다음 명령을 실행해야 했다.game-data-platformgame-data-price즉 구조상으로는 새로운 게임이 Collector 대상에 자동으로 포함됐지만, Collector 자체를 실행하는 것은 여전히 사람이 해야 했다.이번 작업에서는 이 부분을 Airflow와 연결했다.최종적으로 만들고자 한 흐름은 다음과 같다.사용자가 게임 등록 ↓games ↓Airflow Scheduler ↓Player Count Collect..

#18까지 Game Dashboard에서 동접자, 가격, 가격 이력, 최근 패치를 확인할 수 있게 되었다.하지만 한 가지 큰 제약이 있었다.지금까지 Dashboard에서 볼 수 있는 게임은 모두 미리 games 테이블에 등록해둔 게임이었다.즉 새로운 게임을 추가하려면 DB에 직접 데이터를 넣어야 했다.이번 작업에서는 이 구조를 바꿨다.사용자가 Web에서 Steam Store URL 전체를 입력하면:Steam Store URL 입력 ↓Steam App ID 추출 ↓Steam 게임 정보 확인 ↓게임 미리보기 ↓게임 등록 ↓games 테이블 ↓기존 Collector 자동 포함 ↓Game Dashboard까지 이어지는 흐름을..

FastAPI로 API를 만들기 시작하면 처음에는 구조가 단순하다.Route→ DB 조회→ Response예를 들어 특정 게임 정보를 조회하는 API라면 다음 정도로도 충분하다.@router.get("/games/{game_id}")def game_detail(game_id: int): game = get_game(game_id) if game is None: raise HTTPException(status_code=404,detail="Game not found",) return game여기서 get_game()은 DB를 조회하고, Route는 HTTP 요청을 처리한다.구조도 이해하기 쉽다.하지만 서비스가 커지면서 하나의 요청이 여러 작업을 수..

#17에서 Steam 가격 수집 파이프라인을 만들었다.Steam Store API에서 가격을 가져오고, 현재 가격과 가격 변경 이력을 PostgreSQL에 저장할 수 있게 되었지만 데이터가 DB에 쌓이는 것만으로는 사용자가 체감할 수 있는 기능이 되지는 않는다.이번 작업에서는 저장된 가격 데이터를 Game Dashboard에 연결했다.최종적으로 게임 상세 화면에서 다음 정보를 확인할 수 있게 되었다.현재 가격과 할인 여부무료 게임 여부가격을 조회할 수 없는 게임의 상태가격 변경 이력 차트이번 작업에서 특히 중요했던 부분은 단순히 숫자를 화면에 표시하는 것이 아니라 가격이라는 데이터를 어떤 상태로 해석해서 보여줄 것인가였다.1. 가격 데이터는 단순한 숫자가 아니다처음에는 현재 가격만 가져오면 다음처럼 표시..

지난 #17에서는 Steam 가격 데이터를 수집하고, 게임 목록을 코드에 하드코딩하지 않고 DB의 games 테이블을 기준으로 수집하는 구조로 확장했다.이번에는 새로운 데이터 종류를 추가하기보다 기존 패치 파이프라인을 실제로 운영하기 편한 형태로 다듬었다.기존 패치 시스템은 이미 다음 단계까지 구현되어 있었다.공식 패치 소스 수집→ 패치 감지→ game_patches 정규화→ LLM 기반 번역→ LLM 기반 요약→ Web Viewer각 단계는 동작하고 있었지만 중요한 문제가 하나 있었다.패치 수집과 번역·요약 처리가 서로 분리되어 있었다.새로운 패치가 들어오더라도 별도로 번역·요약 Processor를 실행해야 했다.이번 작업에서는 이 흐름을 하나로 연결하고, Web Viewer에서 특정 패치를 직접 다시..