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

#20에서는 기존 Player Count와 Price Collector를 Airflow에 연결하고, 실패 상황에 따른 Skip과 Retry 정책까지 구성했다.

이제 사용자가 게임을 등록하면 Airflow가 정해진 시간에 데이터를 수집하고, 일부 게임에서 오류가 발생하더라도 가능한 범위에서 수집을 계속할 수 있다.

하지만 아직 자동화되지 않은 수집 파이프라인이 하나 남아 있었다.

바로 패치노트 수집 파이프라인이다.

패치노트는 이미 Steam News, PUBG, Overwatch를 대상으로 수집하고 있었고, 새로운 패치를 감지하면 LLM을 통해 한국어 번역과 요약까지 수행할 수 있었다.

문제는 실행 방식이었다.

기존에는 Docker Compose를 시작할 때 별도의 초기화 컨테이너를 통해 패치 수집을 실행하는 구조였다.

이번 작업에서는 기존 수집 코드는 최대한 유지하면서, 패치 수집도 Airflow에서 실행할 수 있도록 연결했다.

---

1. #20까지의 자동화 구조

기존에 Airflow로 관리하던 수집 작업은 두 가지였다.

Player Count
→ 매시간 실행

Price
→ 6시간마다 실행

Player Count는 Dynamic Task Mapping을 사용하여 게임별로 Task를 생성했다.

반면 Price는 하나의 Task에서 등록된 게임을 순회하는 구조였다.

두 방식 모두 실제 수집 로직은 기존 Collector에서 수행하고, Airflow는 실행 시점과 실행 상태를 관리하도록 구성했다.

패치 수집도 같은 원칙을 적용하기로 했다.

Airflow
    ↓
기존 Patch Pipeline
    ↓
Steam / PUBG / Overwatch
    ↓
PostgreSQL
    ↓
LLM 번역 및 요약

새로운 패치 수집기를 작성하는 것이 아니라, 이미 구현된 파이프라인을 Airflow가 호출하도록 만드는 것이 목표였다.

---

2. 기존 Docker 패치 수집 구조 확인

우선 패치 수집이 어디에서 실행되는지 확인했다.

compose.yaml에는 다음 서비스가 있었다.

patch-init:
  build: .
  container_name: game-data-patch-init
  env_file:
    - .env
  environment:
    DB_HOST: db
    DB_PORT: 5432
  depends_on:
    db:
      condition: service_healthy
  command: uv run --no-dev game-data-patches
  restart: "no"

여기서 중요한 것은 다음 명령이다.

uv run --no-dev game-data-patches

pyproject.toml을 확인하니 해당 명령은 다음 함수와 연결되어 있었다.

game-data-patches =
"game_data_platform.patch_pipeline:main"

즉 실행 흐름은 다음과 같았다.

Docker Compose
    ↓
patch-init
    ↓
game-data-patches
    ↓
patch_pipeline.py
    ↓
run_patch_pipeline()

patch-init은 작업을 한 번 수행하고 종료하는 일회성 서비스다.

하지만 정기적인 패치 수집을 관리하는 Scheduler는 아니었다.

서비스가 언제 다시 실행될지는 Docker Compose의 실행 방식과 컨테이너 상태에 영향을 받는다.

따라서 이 구조만으로는 새로운 패치가 발생했는지 일정한 간격으로 확인하기 어렵다.

---

3. 기존 Patch Pipeline은 이미 완성되어 있었다

src/game_data_platform/patch_pipeline.py를 확인했다.

기존 run_patch_pipeline()은 다음 순서로 실행되고 있었다.

1. games 테이블에서 게임 목록 조회

2. Steam News 수집

3. Steam News에서 패치 감지

4. PUBG 공식 패치노트 수집

5. Overwatch 공식 패치노트 수집

6. 공통 game_patches 테이블 동기화

7. 신규/변경 패치 LLM 번역 및 요약

핵심 코드는 다음과 같은 구조였다.

def run_patch_pipeline() -> dict:
    games = get_games()

    steam_collection = collect_news(
        games,
        count=20,
    )

    steam_detection = process_patch_detection()

    pubg_collection = collect_pubg_patch_notes()

    overwatch_collection = (
        collect_overwatch_patch_notes()
    )

    patch_sync = sync_game_patches()

    llm_processing = run_patch_llm_processing(
        process_limit=0,
    )

이미 수집부터 LLM 처리까지 하나의 함수에 연결되어 있었다.

이 함수는 Airflow의 존재를 알 필요가 없다.

직접 실행하든, Docker에서 실행하든, Airflow에서 호출하든 같은 파이프라인을 수행하면 된다.

따라서 Collector 내부를 Airflow 전용 코드로 변경하지 않고 기존 함수를 재사용하기로 했다.

---

4. Airflow에 필요한 Python 패키지 추가

새 DAG를 만들기 전에 실행 환경을 확인했다.

기존 requirements-airflow.txt에는 다음 패키지만 있었다.

httpx>=0.28.1
psycopg[binary]>=3.3.5
python-dotenv>=1.2.3

이 구성으로는 Player Count와 Price Collector를 실행할 수 있었지만, 패치 수집에 필요한 모든 의존성이 포함된 것은 아니었다.

특히 패치 수집과 LLM 처리에는 다음 패키지가 필요했다.

beautifulsoup4
google-genai

그래서 Airflow 의존성에 두 패키지를 추가했다.

httpx>=0.28.1
psycopg[binary]>=3.3.5
python-dotenv>=1.2.3
beautifulsoup4>=4.15.0
google-genai>=2.29.0

여기서 다시 확인한 개념은 Docker 이미지별 실행 환경이 독립적이라는 점이다.

프로젝트의 일반 Python 환경에 라이브러리가 설치되어 있다고 해서 Airflow 이미지에서도 해당 라이브러리를 사용할 수 있는 것은 아니다.

Airflow가 패치 파이프라인을 실행하려면 Airflow 이미지에도 필요한 패키지가 설치되어 있어야 한다.

---

5. Dockerfile 정리 및 이미지 재빌드

Dockerfile.airflow도 확인했다.

기존 파일의 마지막에는 불필요한 줄 연결 문자가 남아 있었다.

이를 정리하고 다음과 같이 유지했다.

FROM apache/airflow:3.3.2-python3.12

COPY requirements-airflow.txt /requirements-airflow.txt

RUN pip install --no-cache-dir \
    "apache-airflow==3.3.2" \
    -r /requirements-airflow.txt

Airflow 버전 자체는 변경하지 않았다.

기존에 정상 동작하던 3.3.2 환경을 유지하면서 필요한 패키지만 추가했다.

이미지는 다음 명령으로 빌드했다.

docker compose build airflow-scheduler

빌드가 정상 완료된 뒤 패치 파이프라인을 import할 수 있는지 테스트했다.

docker compose run --rm --no-deps airflow-scheduler \
  python -c "from game_data_platform.patch_pipeline import run_patch_pipeline; print('Patch pipeline import OK')"

결과는 다음과 같았다.

Patch pipeline import OK

즉 Airflow 이미지에서 기존 패치 파이프라인과 관련 모듈을 정상적으로 불러올 수 있었다.

다만 이 테스트는 함수의 import만 확인한다.

실제 외부 API 호출이나 DB 저장, LLM 실행 성공을 의미하는 것은 아니다.

---

6. Airflow에서 LLM 환경변수 사용하기

패치 파이프라인에는 한 가지 추가 요구사항이 있었다.

바로 LLM 설정이다.

기존 patch-init은 .env를 통해 LLM 설정을 읽고 있었지만, Airflow Scheduler에는 DB 연결 정보만 전달하고 있었다.

현재 프로젝트의 LLM 처리에는 다음 환경변수가 사용된다.

LLM_PROVIDER
LLM_MODEL
GEMINI_API_KEY

기존 Airflow는 LocalExecutor를 사용한다.

따라서 Scheduler가 실행하는 Task 프로세스에서 해당 설정을 사용할 수 있어야 한다.

compose.yaml의 airflow-scheduler 환경변수 영역에 다음 항목을 추가했다.

LLM_PROVIDER: ${LLM_PROVIDER}
LLM_MODEL: ${LLM_MODEL}
GEMINI_API_KEY: ${GEMINI_API_KEY}

여기서 실제 API 키를 코드에 직접 작성하지 않았다.

기존 .env에 정의된 값을 Docker Compose가 참조하도록 했다.

설정 변경 후에는 다음 명령으로 검증했다.

docker compose config --quiet

설정 오류가 없음을 확인한 후, Airflow Scheduler와 DAG Processor를 새 이미지로 재생성했다.

docker compose up -d --no-deps --force-recreate \
  airflow-scheduler \
  airflow-dag-processor

기존 PostgreSQL이나 Redis 컨테이너를 삭제하지 않고 필요한 Airflow 서비스만 교체했다.

실제 Scheduler 내부에서 환경변수의 존재 여부를 검사한 결과 세 항목 모두 정상적으로 전달되어 있었다.

---

7. 새로운 Patch Collection DAG 생성

이제 새로운 DAG 파일을 추가했다.

파일 경로는 다음과 같다.

dags/patch_collection.py

DAG 구성은 기존 Price DAG와 비슷하게 가져갔다.

하지만 이번에는 처음부터 자동 실행하지 않도록 설정했다.

from datetime import timedelta

import pendulum

from airflow.sdk import dag, task

from game_data_platform.patch_pipeline import (
    run_patch_pipeline,
)


@dag(
    dag_id="patch_collection",
    schedule="30 */6 * * *",
    start_date=pendulum.datetime(
        2026,
        1,
        1,
        tz="Asia/Seoul",
    ),
    catchup=False,
    max_active_runs=1,
    is_paused_upon_creation=True,
    tags=[
        "game-data-project",
        "patch",
        "news",
    ],
)
def patch_collection():

    @task(
        retries=0,
    )
    def collect_patches():
        result = run_patch_pipeline(
            llm_process_limit=1,
        )

        print(
            "Patch collection result:",
            result,
        )

    collect_patches()


patch_collection()

여기서 사용한 설정을 정리하면 다음과 같다.

설정의미

schedule="30 */6 * * *"	6시간마다 30분에 실행하도록 정의
catchup=False	과거 스케줄을 한꺼번에 보충하지 않음
max_active_runs=1	동일 DAG의 실행 중첩 제한
is_paused_upon_creation=True	최초 등록 시 자동 실행 방지
retries=0	테스트 중 자동 재시도 방지

패치 수집도 우선 6시간 주기로 설계했다.

Price Collector는 6시간마다 정각에 실행되므로 패치 수집은 30분 차이를 두었다.

현재 프로젝트 규모에서는 패치노트를 수분마다 확인할 필요성이 크지 않고, 외부 API 호출량도 고려해야 한다.

또한 지금은 최초 실행을 검증하는 단계이므로 DAG를 Paused 상태로 유지했다.

---

8. 왜 하나의 Task로 구성했는가

기존 Player Count DAG는 게임별로 Task가 분리되어 있다.

get_games()
    ↓
Game A Task
Game B Task
Game C Task

하지만 이번 Patch Pipeline은 여러 단계가 순서대로 연결되어 있다.

Steam News 수집
    ↓
패치 감지
    ↓
공식 패치 수집
    ↓
공통 테이블 동기화
    ↓
LLM 처리

물론 Airflow에서도 각 단계를 별도의 Task로 나눌 수 있다.

그러면 단계별 재시도나 실행 상태를 분리할 수 있다는 장점이 있다.

하지만 현재는 기존 파이프라인이 이미 하나의 함수로 구성되어 있고, 각 처리 단계도 해당 함수 안에서 순서대로 실행된다.

이를 Airflow에 연결한다는 이유만으로 전부 다시 설계할 필요는 없다고 판단했다.

그래서 이번에는 다음 형태를 선택했다.

Airflow Task
    ↓
run_patch_pipeline()
    ↓
기존 패치 처리 흐름

기존 코드를 재사용하고 자동 실행 연결부터 검증하는 것이 더 단순했다.

---

9. LLM 처리량을 별도 인자로 분리했다

테스트를 준비하면서 한 가지 위험한 부분을 발견했다.

기존 파이프라인에는 다음 코드가 있었다.

run_patch_llm_processing(
    process_limit=0,
)

여기서 0은 처리량을 제한하지 않는다는 의미다.

즉 미처리 패치가 많다면 Airflow 테스트 한 번으로 여러 건의 LLM 요청이 발생할 수 있다.

또한 기존 LLM 처리 명령에서는 환경변수로 처리량을 제어할 수 있었지만, 통합 패치 파이프라인에서는 0을 직접 전달하고 있었다.

그래서 run_patch_pipeline()에 인자를 추가했다.

def run_patch_pipeline(
    llm_process_limit: int = 0,
) -> dict:
    games = get_games()

그리고 LLM 호출 부분을 다음과 같이 변경했다.

llm_processing = run_patch_llm_processing(
    process_limit=llm_process_limit,
)

이제 기존 명령은 기본값을 사용하고, Airflow에서는 처리량을 별도로 지정할 수 있다.

이번 테스트에는 다음 값을 적용했다.

llm_process_limit=1

한 번의 파이프라인 실행에서 LLM 처리 시도 건수를 최대 1건으로 제한한 것이다.

또한 초기 테스트에서는 Airflow Retry를 0으로 설정했다.

실제 외부 API와 DB를 사용하는 작업이므로, 검증이 끝나기 전 자동 재시도에 의해 같은 파이프라인이 반복 실행되는 상황을 피하기 위해서다.

---

10. DAG 등록 및 Import 검증

새 파일을 작성한 뒤 먼저 Python 문법을 확인했다.

python -m py_compile \
  src/game_data_platform/patch_pipeline.py \
  dags/patch_collection.py

문법 오류는 발생하지 않았다.

이후 Airflow의 DAG 목록과 Import Error를 확인했다.

docker compose exec airflow-dag-processor \
  airflow dags list

docker compose exec airflow-dag-processor \
  airflow dags list-import-errors

새로운 patch_collection DAG가 등록되었고, Import Error는 발생하지 않았다.

No data found

Airflow UI에서도 DAG 등록과 Paused 상태를 확인했다.

이제 DAG 자체를 Airflow가 정상적으로 인식한다는 것까지 확인한 것이다.

하지만 DAG가 등록됐다고 해서 실제 수집이 성공했다는 의미는 아니다.

마지막으로 전체 파이프라인을 실행해 봐야 한다.

---

11. Airflow에서 실제 패치 수집 테스트

이번에는 정기 스케줄을 활성화하지 않고 DAG 테스트 기능을 사용했다.

docker compose exec airflow-scheduler \
  airflow dags test patch_collection

이 명령은 DAG를 한 번 실행하여 실제 Task 동작을 확인할 수 있다.

테스트에서는 기존 데이터베이스와 외부 API를 그대로 사용했다.

실행 결과는 다음과 같았다.

collect_patches
→ SUCCESS

patch_collection
→ SUCCESS

실행 시간
→ 약 13초

즉 Airflow에서 기존 패치 파이프라인을 정상적으로 호출하고 실행을 마쳤다.

하지만 Airflow의 SUCCESS는 Task 함수가 예외 없이 종료됐다는 의미이므로, 내부 Collector의 결과도 확인해야 한다.

---

12. Steam News 수집 결과

이번 실행에서는 DB에 등록된 18개 게임을 대상으로 Steam News를 조회했다.

게임별로 최근 20건을 가져오도록 설정되어 있었다.

등록된 게임: 18개

게임별 조회: 20건

신규 뉴스 INSERT: 1건

대부분의 게임은 이미 저장된 뉴스만 조회되었으므로 신규 INSERT가 발생하지 않았다.

새로운 뉴스가 발견된 게임은 다음과 같았다.

콜 오브 듀티: 모던 워페어 4

fetched: 20
inserted: 1

새 뉴스 1건에 대해 패치 감지 로직이 실행되었다.

total: 1
patch: 0
non_patch: 1
detector_version: v2

해당 뉴스는 패치노트가 아닌 일반 뉴스로 분류되었다.

이번 테스트를 통해 Airflow에서도 Steam News 수집과 패치 감지 흐름이 실행되는 것을 확인할 수 있었다.

---

13. PUBG와 Overwatch 수집 결과

공식 패치 소스도 정상적으로 호출되었다.

PUBG

candidates: 1
collected: 0
unchanged: 1
failed: 0

패치 후보 1건을 확인했지만 기존 데이터와 동일하여 새롭게 수집할 항목은 없었다.

실패도 발생하지 않았다.

Overwatch

fetched: 6
inserted: 0

ko-KR
→ fetched: 3

en-US
→ fetched: 3

한국어와 영어 패치노트를 각각 3건씩 조회했다.

모두 기존 데이터였으므로 신규 INSERT는 발생하지 않았다.

이번 실행에서는 Steam, PUBG, Overwatch Collector를 하나의 Airflow Task 안에서 호출하는 데 성공했다.

---

14. 공통 패치 동기화와 LLM 결과

수집 후 sync_game_patches()가 실행되었다.

결과는 다음과 같았다.

steam_synced: 0
pubg_synced: 0
overwatch_synced: 0
total_synced: 0

이번에 발견된 새로운 Steam News는 패치가 아닌 일반 뉴스였고, PUBG와 Overwatch에서도 새로운 패치 데이터가 없었다.

따라서 공통 패치 테이블에 새롭게 동기화된 데이터가 없었다.

이후 LLM 처리 결과는 다음과 같았다.

llm_processing:
    processed: 0

이번 실행에서는 번역·요약이 필요한 신규 또는 변경 패치가 없었기 때문에 LLM 처리 대상도 없었다.

이는 테스트 실패가 아니다.

다만 이번 테스트로 실제 LLM API 호출과 번역·요약 저장까지 검증한 것은 아니다.

확인한 것은 Airflow가 통합 파이프라인을 실행하고 LLM 처리 단계까지 정상적으로 도달했다는 점이다.

---

15. 로그가 두 번 출력된 이유

실행 로그를 확인하면서 Patch collection result가 두 번 출력되는 것을 발견했다.

처음에는 파이프라인이 중복 실행된 것처럼 보일 수도 있다.

하지만 원인은 단순했다.

patch_pipeline.py에는 다음 로그가 있었다.

logger.info(
    "Patch pipeline result: %s",
    result,
)

그리고 Airflow DAG에서도 결과를 출력하고 있었다.

print(
    "Patch collection result:",
    result,
)

따라서 동일한 결과가 두 번 출력된 것이다.

수집이 두 번 실행된 것은 아니었다.

현재는 기능상 문제가 없으므로 그대로 유지했다.

---

16. 기존 patch-init은 수동 실행 전용으로 변경

Airflow 연결이 정상적으로 확인됐으므로 기존 Docker 실행 방식도 정리했다.

문제는 Docker 시작 시 기존 patch-init이 실행되고, Airflow에서도 정기 수집을 수행하게 되면 실행 경로가 중복될 수 있다는 점이었다.

그래서 patch-init을 완전히 삭제하는 대신 Docker Compose의 Profile 기능을 사용했다.

기존 서비스에 다음 설정을 추가했다.

patch-init:
  profiles: ["manual"]

Profile을 사용하면 일반적인 docker compose up -d 실행 대상에서 해당 서비스를 제외할 수 있다.

수정 후 설정을 검증했다.

docker compose config --quiet

그리고 기본 서비스 목록을 확인했다.

docker compose config --services

출력 결과에 patch-init은 포함되지 않았다.

즉 앞으로 기본 Compose 실행에서는 해당 초기화 서비스를 자동으로 실행하지 않는다.

다만 필요할 때 수동 실행은 가능하다.

docker compose run --rm patch-init

이 방식은 기존 수집 명령을 비상용 또는 수동 테스트 용도로 남겨두면서, 자동 실행의 관리 주체를 Airflow로 통일할 수 있다는 장점이 있다.

---

17. 이번 작업에서 책임이 어떻게 달라졌는가

기존 구조는 다음과 같았다.

Docker Compose
    ↓
patch-init
    ↓
Patch Pipeline
    ↓
PostgreSQL

이번에 Airflow와 연결한 구조는 다음과 같다.

Airflow Scheduler
    ↓
patch_collection DAG
    ↓
collect_patches Task
    ↓
run_patch_pipeline()
    ↓
Steam / PUBG / Overwatch
    ↓
game_patches
    ↓
LLM 번역 및 요약

그리고 Docker Compose는 다음 역할을 유지한다.

Docker Compose
→ 컨테이너 실행 환경 관리

Airflow
→ 실행 시점과 Task 상태 관리

Patch Pipeline
→ 실제 수집 및 처리 로직

PostgreSQL
→ 수집 결과와 처리 상태 저장

이번 작업은 새로운 수집 기능을 개발한 것이 아니라 기존 구성 요소들의 실행 책임을 더 명확하게 나누는 작업이었다.

---

18. 이번 작업의 완료 범위

이번 #21-1에서 완료한 항목은 다음과 같다.

항목결과

기존 Patch Pipeline 구조 확인	완료
Airflow 패키지 의존성 보완	완료
Airflow LLM 환경변수 연결	완료
Patch Collection DAG 추가	완료
DAG Import 검증	완료
실제 수집 파이프라인 테스트	완료
Docker patch-init 기본 실행 제외	완료
Git 커밋 및 원격 푸시	완료

Git에는 다음 커밋으로 정리했다.

736174f
feat: integrate patch collection pipeline with Airflow

다만 이번 작업이 끝난 시점에도 DAG는 Paused 상태로 유지했다.

또한 테스트용으로 다음 값을 사용하고 있다.

retries=0
llm_process_limit=1

따라서 아직 정기 실행을 활성화한 상태는 아니다.

운영용 재시도 정책과 LLM 처리량을 결정하고 정기 실행을 활성화하는 작업은 별도로 검증할 필요가 있다.

---

19. #21-1을 마치며

이번 작업에서는 이미 구현된 패치 수집 기능을 Airflow에서 실행할 수 있도록 연결했다.

처음에는 새로운 DAG 파일 하나만 작성하면 끝날 것으로 생각할 수 있었다.

하지만 실제로 진행해 보니 다음 사항을 함께 고려해야 했다.

Airflow 이미지에 필요한 패키지가 있는가?

LLM 환경변수가 Task에 전달되는가?

기존 Collector를 그대로 재사용할 수 있는가?

테스트 중 과도한 LLM 요청이 발생하지 않는가?

기존 Docker 실행과 중복되지 않는가?

Airflow SUCCESS가 실제 수집 성공을 의미하는가?

단순히 Python 함수를 Airflow에서 호출하는 것과 실제로 운영할 수 있는 구조를 만드는 것 사이에는 차이가 있었다.

특히 기존 Collector를 다시 구현하지 않고 재사용했다는 점이 중요했다.

#20에서 Player Count와 Price Collector를 자동화한 것과 같은 방식으로, 이번에는 Patch Pipeline도 Airflow가 실행할 수 있는 구조를 만들었다.

이제 기본 수집 시스템은 다음 세 영역으로 확장되었다.

Player Count
→ Airflow 정기 수집

Price
→ Airflow 정기 수집

Patch
→ Airflow DAG 연결 및 실행 검증 완료

패치 수집의 정기 실행은 아직 활성화하지 않았지만, 실제 통합 파이프라인이 Airflow에서 정상적으로 실행되는 것까지 확인했다.

이번 작업을 통해 다시 느낀 점은 자동화에서는 작업을 실행하는 것뿐 아니라, 실행 환경과 실행 주체, 실패 상태를 어떻게 관리할지도 중요하다는 것이다.

기존 기능을 최대한 유지하면서 운영 구조를 한 단계 확장한 작업이었다.