🤖

본 콘텐츠의 이미지 및 내용은 AI로 생성되었습니다.

⚠️

본 콘텐츠의 이미지 및 내용을 무단으로 복제, 배포, 수정하여 사용할 경우 저작권법에 의해 법적 제재를 받을 수 있습니다.

이미지 로딩 중...

종합 프로젝트 트래픽 예측 시스템 구축 - 슬라이드 1/12
A

AI Generated

2025. 12. 3. · 17 Views

종합 프로젝트 트래픽 예측 시스템 구축

실무에서 바로 활용할 수 있는 트래픽 예측 시스템을 처음부터 끝까지 구축하는 종합 프로젝트입니다. 데이터 수집부터 모델 학습, API 배포까지 전체 파이프라인을 단계별로 학습합니다.


목차

  1. 프로젝트_아키텍처_설계
  2. 시계열_데이터_수집
  3. 데이터_전처리_파이프라인
  4. 시계열_특성_추출
  5. 시계열_예측_모델_구현
  6. 모델_평가_지표_설계
  7. 예측_API_서버_구축
  8. 배치_예측_시스템_구현
  9. 모델_모니터링_시스템
  10. 자동_재학습_파이프라인
  11. 종합_테스트_및_배포

1. 프로젝트 아키텍처 설계

김개발 씨는 이번에 처음으로 데이터 사이언스 프로젝트를 리드하게 되었습니다. "트래픽 예측 시스템을 만들어주세요"라는 요청을 받았는데, 막상 시작하려니 어디서부터 손을 대야 할지 막막했습니다.

박시니어 씨가 다가와 말합니다. "일단 전체 그림부터 그려봐요."

프로젝트 아키텍처는 시스템의 전체 구조를 설계하는 것입니다. 마치 건축가가 집을 짓기 전에 설계도를 그리는 것과 같습니다.

데이터 수집, 전처리, 모델 학습, 예측, 배포까지 각 단계가 어떻게 연결되는지 먼저 파악해야 합니다.

다음 코드를 살펴봅시다.

# 트래픽 예측 시스템 아키텍처 설정
class TrafficPredictionSystem:
    def __init__(self, config_path: str):
        # 설정 파일 로드
        self.config = self._load_config(config_path)
        # 각 컴포넌트 초기화
        self.data_collector = DataCollector(self.config['data'])
        self.preprocessor = DataPreprocessor(self.config['preprocessing'])
        self.model = TrafficModel(self.config['model'])
        self.api_server = PredictionAPI(self.config['api'])

    def run_pipeline(self):
        # 전체 파이프라인 실행
        raw_data = self.data_collector.collect()
        processed_data = self.preprocessor.transform(raw_data)
        self.model.train(processed_data)
        return self.model.evaluate()

김개발 씨는 입사 1년 차 데이터 엔지니어입니다. 어느 날 팀장님이 다가와 말씀하셨습니다.

"우리 서비스 트래픽을 예측하는 시스템을 만들어야 하는데, 김개발 씨가 한번 맡아볼래요?" 갑작스러운 제안에 김개발 씨는 당황했습니다. 머신러닝 모델은 학교에서 배웠지만, 실제 시스템을 처음부터 끝까지 구축해본 적은 없었기 때문입니다.

점심시간, 사내 카페에서 박시니어 씨를 만났습니다. 고민을 털어놓자 박시니어 씨가 말했습니다.

"처음 하는 프로젝트에서 가장 중요한 건 뭔지 알아요? 바로 전체 그림을 먼저 그리는 거예요." 프로젝트 아키텍처란 무엇일까요?

쉽게 비유하자면, 이것은 마치 여행 계획표와 같습니다. 어디서 출발해서 어디를 거쳐 최종 목적지까지 가는지 미리 그려놓는 것입니다.

이 계획표가 없으면 중간에 길을 잃거나 불필요한 곳을 돌아가게 됩니다. 트래픽 예측 시스템의 아키텍처는 크게 네 단계로 나뉩니다.

첫째, 데이터 수집 단계에서는 로그 파일이나 데이터베이스에서 원본 데이터를 가져옵니다. 둘째, 전처리 단계에서는 수집한 데이터를 모델이 이해할 수 있는 형태로 변환합니다.

셋째, 모델 학습 단계에서는 전처리된 데이터로 예측 모델을 훈련시킵니다. 넷째, 배포 단계에서는 학습된 모델을 API로 서빙하여 실시간 예측을 제공합니다.

위 코드를 살펴보면, TrafficPredictionSystem 클래스가 전체 시스템의 중심 역할을 합니다. 생성자에서 설정 파일을 읽고 각 컴포넌트를 초기화합니다.

run_pipeline 메서드는 데이터 수집부터 모델 평가까지 전체 흐름을 순차적으로 실행합니다. 왜 이렇게 모듈화된 구조가 중요할까요?

예를 들어 나중에 데이터 소스가 바뀌더라도 DataCollector만 수정하면 됩니다. 모델을 바꾸고 싶다면 TrafficModel만 교체하면 됩니다.

각 부품이 독립적으로 동작하기 때문에 유지보수가 훨씬 수월해집니다. 실무에서는 이 아키텍처를 팀원들과 공유하고 리뷰를 받는 과정이 필수입니다.

김개발 씨도 박시니어 씨의 조언을 받아 먼저 설계 문서를 작성했습니다. 덕분에 나중에 발생할 수 있는 많은 문제를 사전에 방지할 수 있었습니다.

실전 팁

💡 - 설계 단계에서 충분한 시간을 투자하면 구현 단계에서 시간을 절약할 수 있습니다

  • 각 컴포넌트는 단일 책임 원칙을 따라 하나의 역할만 담당하도록 설계하세요
  • 설정 파일을 분리하면 환경별로 다른 설정을 쉽게 적용할 수 있습니다

2. 시계열 데이터 수집

아키텍처를 설계한 김개발 씨는 이제 실제 데이터를 수집해야 합니다. 그런데 트래픽 데이터는 일반 데이터와 다른 특성이 있었습니다.

시간에 따라 흐르는 데이터, 즉 시계열 데이터라는 것이었습니다.

시계열 데이터는 시간 순서에 따라 기록된 데이터입니다. 마치 일기장처럼 매일매일의 기록이 쌓여서 하나의 흐름을 만듭니다.

트래픽 예측에서는 과거의 패턴을 분석하여 미래를 예측하기 때문에 시간 정보가 핵심입니다.

다음 코드를 살펴봅시다.

import pandas as pd
from datetime import datetime, timedelta

class DataCollector:
    def __init__(self, config: dict):
        self.source = config['source']
        self.time_column = config.get('time_column', 'timestamp')

    def collect(self, start_date: str, end_date: str) -> pd.DataFrame:
        # 데이터베이스에서 트래픽 로그 조회
        query = f"""
            SELECT timestamp, request_count, response_time, error_rate
            FROM traffic_logs
            WHERE timestamp BETWEEN '{start_date}' AND '{end_date}'
            ORDER BY timestamp
        """
        df = pd.read_sql(query, self.source)
        # 시간 인덱스 설정
        df[self.time_column] = pd.to_datetime(df[self.time_column])
        df.set_index(self.time_column, inplace=True)
        return df

김개발 씨가 첫 번째로 마주한 도전은 데이터 수집이었습니다. 회사의 데이터베이스에는 수백만 건의 트래픽 로그가 쌓여 있었습니다.

하지만 이 데이터를 어떻게 가져와야 할지 막막했습니다. 박시니어 씨가 힌트를 주었습니다.

"트래픽 데이터는 시계열 데이터예요. 시간의 흐름을 반드시 기억해야 해요." 시계열 데이터란 무엇일까요?

쉽게 비유하자면, 이것은 마치 주식 차트와 같습니다. 오늘의 주가는 어제의 주가와 연결되어 있고, 어제의 주가는 그저께와 연결되어 있습니다.

이런 시간적 연속성이 바로 시계열 데이터의 핵심입니다. 트래픽 데이터도 마찬가지입니다.

오후 2시의 트래픽은 오후 1시의 트래픽과 관련이 있습니다. 또한 오늘 오후 2시의 트래픽은 어제 같은 시간의 트래픽과도 패턴이 유사할 수 있습니다.

이런 시간적 패턴을 분석하는 것이 트래픽 예측의 핵심입니다. 위 코드에서 DataCollector 클래스를 살펴보겠습니다.

collect 메서드는 시작일과 종료일을 받아 해당 기간의 트래픽 로그를 조회합니다. 여기서 중요한 것은 ORDER BY timestamp입니다.

시계열 데이터는 반드시 시간 순서대로 정렬되어야 합니다. 데이터를 가져온 후에는 pd.to_datetime으로 타임스탬프를 날짜 형식으로 변환합니다.

그리고 set_index로 타임스탬프를 데이터프레임의 인덱스로 설정합니다. 이렇게 하면 나중에 시간 기반 분석이 훨씬 편리해집니다.

실무에서는 데이터 수집 시 몇 가지 주의할 점이 있습니다. 첫째, 누락된 시간대가 없는지 확인해야 합니다.

둘째, 타임존을 통일해야 합니다. 서버마다 다른 타임존을 사용하면 데이터가 엉망이 될 수 있습니다.

셋째, 충분한 기간의 데이터를 수집해야 합니다. 보통 최소 몇 주에서 몇 달 치 데이터가 필요합니다.

김개발 씨는 3개월 치 트래픽 데이터를 수집했습니다. 데이터를 확인해보니 약 800만 건의 로그가 있었습니다.

이제 이 데이터를 전처리할 차례입니다.

실전 팁

💡 - 시계열 데이터는 반드시 시간 순서대로 정렬하세요

  • 타임존을 UTC로 통일하면 혼란을 방지할 수 있습니다
  • 최소 2-3개월 이상의 데이터를 수집하여 계절성 패턴을 파악하세요

3. 데이터 전처리 파이프라인

수집한 데이터를 살펴보던 김개발 씨는 당황했습니다. 중간중간 비어있는 값도 있고, 이상하게 튀는 숫자도 있었습니다.

"이 데이터를 그대로 모델에 넣으면 안 될 것 같은데..." 박시니어 씨가 고개를 끄덕입니다. "맞아요, 전처리가 필요해요."

데이터 전처리는 원본 데이터를 모델이 학습하기 좋은 형태로 변환하는 과정입니다. 마치 요리 전에 재료를 손질하는 것과 같습니다.

아무리 좋은 재료도 손질을 잘못하면 요리가 망가지듯, 전처리를 제대로 하지 않으면 모델 성능이 떨어집니다.

다음 코드를 살펴봅시다.

import numpy as np
from sklearn.preprocessing import StandardScaler

class DataPreprocessor:
    def __init__(self, config: dict):
        self.scaler = StandardScaler()
        self.resample_freq = config.get('resample_freq', '1H')

    def transform(self, df: pd.DataFrame) -> pd.DataFrame:
        # 1시간 단위로 리샘플링
        df_resampled = df.resample(self.resample_freq).mean()
        # 결측값 처리 - 선형 보간법 사용
        df_filled = df_resampled.interpolate(method='linear')
        # 이상치 제거 - IQR 방법
        Q1, Q3 = df_filled.quantile(0.25), df_filled.quantile(0.75)
        IQR = Q3 - Q1
        df_clean = df_filled[~((df_filled < Q1 - 1.5*IQR) |
                               (df_filled > Q3 + 1.5*IQR)).any(axis=1)]
        # 정규화
        df_scaled = pd.DataFrame(self.scaler.fit_transform(df_clean),
                                 index=df_clean.index, columns=df_clean.columns)
        return df_scaled

김개발 씨가 수집한 800만 건의 데이터를 자세히 들여다보았습니다. 문제가 한두 가지가 아니었습니다.

어떤 시간대는 데이터가 1초 단위로 있고, 어떤 시간대는 10분 단위로 있었습니다. 중간에 서버 장애로 몇 시간 동안 데이터가 없는 구간도 있었습니다.

또한 간혹 트래픽이 평소의 100배가 넘는 이상한 값도 보였습니다. 박시니어 씨가 말했습니다.

"이런 데이터를 그대로 모델에 넣으면 어떻게 될 것 같아요?" 김개발 씨가 대답했습니다. "아마 모델이 제대로 학습하지 못할 것 같습니다." 정답입니다.

데이터 전처리는 왜 필요할까요? 쉽게 비유하자면, 이것은 마치 요리 전 재료 손질과 같습니다.

싱싱한 야채도 흙이 묻어 있으면 씻어야 하고, 상한 부분은 잘라내야 합니다. 크기도 비슷하게 썰어야 골고루 익습니다.

데이터도 마찬가지입니다. 위 코드의 transform 메서드는 네 단계의 전처리를 수행합니다.

첫 번째 단계는 리샘플링입니다. 불규칙한 간격의 데이터를 1시간 단위로 통일합니다.

resample('1H').mean()은 1시간 동안의 평균값을 계산합니다. 이렇게 하면 데이터의 간격이 일정해집니다.

두 번째 단계는 결측값 처리입니다. interpolate(method='linear')는 선형 보간법을 사용합니다.

예를 들어 10시에 100, 12시에 200이 있고 11시가 비어있다면 150으로 채웁니다. 마치 점과 점 사이를 직선으로 연결하는 것과 같습니다.

세 번째 단계는 이상치 제거입니다. IQR(사분위 범위) 방법을 사용합니다.

데이터의 중간 50%를 기준으로 너무 높거나 낮은 값을 제거합니다. 서버 장애나 DDoS 공격으로 인한 비정상적인 트래픽을 걸러낼 수 있습니다.

네 번째 단계는 정규화입니다. StandardScaler는 데이터를 평균 0, 표준편차 1로 변환합니다.

이렇게 하면 서로 다른 스케일의 변수들을 동일한 기준으로 비교할 수 있습니다. 김개발 씨는 전처리를 마친 후 데이터를 다시 확인했습니다.

깔끔하게 정리된 데이터를 보니 마음이 놓였습니다.

실전 팁

💡 - 전처리 전후의 데이터 분포를 시각화하여 확인하세요

  • fit_transform은 학습 데이터에만 사용하고, 테스트 데이터에는 transform만 사용해야 합니다
  • 이상치를 무조건 제거하지 말고, 원인을 먼저 파악하세요

4. 시계열 특성 추출

전처리를 마친 김개발 씨에게 박시니어 씨가 물었습니다. "혹시 요일별로 트래픽 패턴이 다를 것 같지 않아요?

월요일 아침과 일요일 저녁의 트래픽이 같을까요?" 김개발 씨는 무릎을 쳤습니다. 시간 정보에서 더 많은 특성을 뽑아낼 수 있다는 것을 깨달았습니다.

특성 추출은 원본 데이터에서 모델 학습에 유용한 정보를 추출하는 과정입니다. 시계열 데이터에서는 시간 정보를 활용하여 요일, 시간대, 주말 여부 등의 특성을 만들 수 있습니다.

또한 과거 값을 활용한 지연 특성도 중요합니다.

다음 코드를 살펴봅시다.

class FeatureExtractor:
    def __init__(self, lag_periods: list = [1, 24, 168]):
        self.lag_periods = lag_periods  # 1시간, 1일, 1주 전

    def extract(self, df: pd.DataFrame) -> pd.DataFrame:
        result = df.copy()
        # 시간 기반 특성 추출
        result['hour'] = result.index.hour
        result['day_of_week'] = result.index.dayofweek
        result['is_weekend'] = result.index.dayofweek >= 5
        result['month'] = result.index.month
        # 지연 특성 생성 (과거 값 활용)
        for lag in self.lag_periods:
            result[f'traffic_lag_{lag}'] = result['request_count'].shift(lag)
        # 이동 평균 특성
        result['rolling_mean_24h'] = result['request_count'].rolling(24).mean()
        result['rolling_std_24h'] = result['request_count'].rolling(24).std()
        # 결측값 제거 후 반환
        return result.dropna()

김개발 씨는 전처리된 데이터를 시각화해보았습니다. 놀랍게도 뚜렷한 패턴이 보였습니다.

평일 오전 9시부터 트래픽이 급증하고, 점심시간에 살짝 줄었다가, 오후에 다시 늘어났습니다. 반면 주말에는 전체적으로 트래픽이 낮았습니다.

박시니어 씨가 말했습니다. "이런 패턴을 모델에게 알려줘야 해요.

지금 데이터에는 타임스탬프만 있잖아요. 모델이 월요일과 일요일의 차이를 알 수 있을까요?" 특성 추출이란 무엇일까요?

쉽게 비유하자면, 이것은 마치 탐정이 단서를 찾는 것과 같습니다. 범인을 잡으려면 현장에서 최대한 많은 단서를 수집해야 합니다.

발자국, 지문, 목격자 증언 등 다양한 정보가 필요합니다. 데이터에서도 마찬가지로 모델이 예측하는 데 도움이 될 다양한 단서를 추출해야 합니다.

위 코드에서 FeatureExtractor 클래스를 살펴보겠습니다. extract 메서드는 크게 세 종류의 특성을 생성합니다.

첫 번째는 시간 기반 특성입니다. hour는 0부터 23까지의 시간대를 나타냅니다.

day_of_week는 요일을 숫자로 표현합니다 (0은 월요일, 6은 일요일). is_weekend는 주말 여부를 True/False로 나타냅니다.

month는 계절성을 파악하는 데 유용합니다. 두 번째는 지연 특성입니다.

traffic_lag_1은 1시간 전의 트래픽 값입니다. traffic_lag_24는 어제 같은 시간의 트래픽입니다.

traffic_lag_168은 일주일 전 같은 시간의 트래픽입니다. 이런 지연 특성이 중요한 이유는 "지금의 트래픽은 과거의 트래픽과 연관이 있다"는 시계열 데이터의 특성 때문입니다.

세 번째는 이동 통계 특성입니다. rolling_mean_24h는 최근 24시간의 평균 트래픽입니다.

rolling_std_24h는 최근 24시간의 트래픽 변동성을 나타냅니다. 이런 특성은 최근 추세를 파악하는 데 도움이 됩니다.

마지막으로 dropna()를 호출하여 지연 특성 생성 과정에서 발생한 결측값을 제거합니다. 예를 들어 데이터의 첫 168개 행은 일주일 전 값이 없기 때문에 제거됩니다.

김개발 씨는 특성을 추출한 후 데이터의 컬럼 수가 크게 늘어난 것을 확인했습니다. 이제 모델이 더 풍부한 정보를 활용할 수 있게 되었습니다.

실전 팁

💡 - 도메인 지식을 활용하여 의미 있는 특성을 추가하세요 (예: 공휴일 여부)

  • 지연 특성의 기간은 데이터의 주기성을 고려하여 선택하세요
  • 너무 많은 특성은 오버피팅을 유발할 수 있으니 주의하세요

5. 시계열 예측 모델 구현

드디어 모델을 만들 차례가 왔습니다. 김개발 씨는 LSTM, Prophet, XGBoost 등 다양한 모델 이름을 들어보았지만 어떤 것을 선택해야 할지 몰랐습니다.

박시니어 씨가 조언합니다. "처음에는 간단한 모델부터 시작해요.

성능을 확인한 후에 점진적으로 복잡한 모델을 시도해보는 거예요."

시계열 예측 모델은 과거 데이터의 패턴을 학습하여 미래 값을 예측합니다. 처음에는 LightGBM 같은 트리 기반 모델로 시작하는 것이 좋습니다.

구현이 간단하고 성능도 우수하며, 특성 중요도를 파악할 수 있어 모델을 이해하기 쉽습니다.

다음 코드를 살펴봅시다.

import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit

class TrafficModel:
    def __init__(self, config: dict):
        self.params = config.get('params', {
            'objective': 'regression',
            'metric': 'rmse',
            'num_leaves': 31,
            'learning_rate': 0.05
        })
        self.model = None

    def train(self, df: pd.DataFrame, target_col: str = 'request_count'):
        X = df.drop(columns=[target_col])
        y = df[target_col]
        # 시계열 교차 검증 사용
        tscv = TimeSeriesSplit(n_splits=5)
        train_idx, val_idx = list(tscv.split(X))[-1]
        train_data = lgb.Dataset(X.iloc[train_idx], y.iloc[train_idx])
        val_data = lgb.Dataset(X.iloc[val_idx], y.iloc[val_idx])
        # 모델 학습
        self.model = lgb.train(self.params, train_data,
                               valid_sets=[val_data], num_boost_round=1000,
                               callbacks=[lgb.early_stopping(50)])
        return self

김개발 씨는 모델 선택에 대해 고민이 깊었습니다. 인터넷을 검색해보니 LSTM이 가장 좋다는 글도 있고, Prophet이 간편하다는 글도 있었습니다.

어떤 글에서는 XGBoost가 최고라고 했습니다. 박시니어 씨가 말했습니다.

"모델 선택에는 정답이 없어요. 중요한 건 우리 데이터에 맞는 모델을 찾는 거예요.

일단 LightGBM으로 시작해봐요. 빠르고 성능도 좋거든요." LightGBM은 그래디언트 부스팅 기반의 머신러닝 모델입니다.

쉽게 비유하자면, 이것은 마치 여러 명의 전문가가 함께 예측하는 것과 같습니다. 한 전문가가 틀린 부분을 다음 전문가가 보완하고, 그 다음 전문가가 또 보완합니다.

이렇게 여러 전문가의 의견을 종합하면 더 정확한 예측이 가능합니다. 위 코드에서 가장 중요한 부분은 TimeSeriesSplit입니다.

일반적인 교차 검증은 데이터를 무작위로 섞어서 나눕니다. 하지만 시계열 데이터에서는 이렇게 하면 안 됩니다.

왜냐하면 미래 데이터가 학습에 포함될 수 있기 때문입니다. 예를 들어 1월부터 12월까지의 데이터가 있다고 가정해봅시다.

무작위로 섞으면 12월 데이터로 1월을 예측하는 상황이 생길 수 있습니다. 하지만 실제로는 12월 데이터를 미리 알 수 없습니다.

TimeSeriesSplit은 항상 과거 데이터로 학습하고 미래 데이터로 검증합니다. 모델 파라미터를 살펴보겠습니다.

num_leaves는 트리의 복잡도를 결정합니다. 값이 클수록 복잡한 패턴을 학습할 수 있지만 오버피팅 위험도 높아집니다.

learning_rate는 학습 속도를 조절합니다. 값이 작을수록 천천히 학습하지만 더 정교한 모델을 만들 수 있습니다.

early_stopping 콜백은 과적합을 방지합니다. 검증 데이터의 성능이 50라운드 동안 개선되지 않으면 학습을 조기 종료합니다.

이렇게 하면 불필요한 학습을 줄이고 최적의 성능을 유지할 수 있습니다. 김개발 씨는 모델을 학습시킨 후 특성 중요도를 확인했습니다.

예상대로 traffic_lag_24와 hour가 가장 중요한 특성으로 나타났습니다.

실전 팁

💡 - 시계열 데이터에서는 반드시 TimeSeriesSplit을 사용하세요

  • 처음에는 간단한 모델로 베이스라인을 설정한 후 점진적으로 개선하세요
  • 특성 중요도를 분석하여 모델의 예측 근거를 이해하세요

6. 모델 평가 지표 설계

모델 학습을 마친 김개발 씨가 결과를 보고하러 갔습니다. "RMSE가 150입니다!"라고 말했는데, 팀장님이 고개를 갸우뚱합니다.

"그게 좋은 건가요, 나쁜 건가요?" 김개발 씨는 대답하지 못했습니다. 숫자만 봐서는 모델이 얼마나 좋은지 알 수 없었습니다.

모델 평가 지표는 모델의 성능을 측정하는 기준입니다. 트래픽 예측에서는 RMSE, MAE, MAPE 등 다양한 지표를 사용합니다.

하나의 지표만 보면 전체 그림을 놓칠 수 있으므로 여러 지표를 함께 확인해야 합니다.

다음 코드를 살펴봅시다.

from sklearn.metrics import mean_squared_error, mean_absolute_error
import numpy as np

class ModelEvaluator:
    def __init__(self, model, scaler=None):
        self.model = model
        self.scaler = scaler

    def evaluate(self, X_test: pd.DataFrame, y_test: pd.Series) -> dict:
        y_pred = self.model.predict(X_test)
        # 스케일러가 있으면 원래 스케일로 복원
        if self.scaler:
            y_test = self.scaler.inverse_transform(y_test.values.reshape(-1, 1))
            y_pred = self.scaler.inverse_transform(y_pred.reshape(-1, 1))
        # 다양한 평가 지표 계산
        rmse = np.sqrt(mean_squared_error(y_test, y_pred))
        mae = mean_absolute_error(y_test, y_pred)
        mape = np.mean(np.abs((y_test - y_pred) / y_test)) * 100
        return {
            'RMSE': rmse,      # 제곱근 평균 제곱 오차
            'MAE': mae,        # 평균 절대 오차
            'MAPE': mape       # 평균 절대 백분율 오차
        }

박시니어 씨가 김개발 씨에게 물었습니다. "RMSE 150이라고 했는데, 우리 서비스의 평균 트래픽이 얼마예요?" 김개발 씨가 확인해보니 평균 트래픽은 약 10,000건이었습니다.

"그러면 오차가 평균의 1.5% 정도네요. 나쁘지 않은 것 같은데요?" 모델 평가 지표는 왜 중요할까요?

쉽게 비유하자면, 이것은 마치 건강검진 결과와 같습니다. 혈압만 보면 건강한지 알 수 없습니다.

혈당, 콜레스테롤, 간 수치 등 다양한 지표를 함께 봐야 전체적인 건강 상태를 파악할 수 있습니다. 위 코드에서 세 가지 평가 지표를 계산합니다.

RMSE(Root Mean Squared Error)는 가장 널리 사용되는 지표입니다. 예측값과 실제값의 차이를 제곱하여 평균을 낸 후 제곱근을 취합니다.

큰 오차에 더 민감하다는 특징이 있습니다. 만약 한 번의 큰 오차가 치명적인 상황이라면 RMSE를 중요하게 봐야 합니다.

MAE(Mean Absolute Error)는 예측값과 실제값의 차이의 절대값 평균입니다. RMSE보다 이상치에 덜 민감합니다.

평균적으로 얼마나 틀리는지를 직관적으로 보여줍니다. MAPE(Mean Absolute Percentage Error)는 백분율로 표현된 오차입니다.

"평균적으로 5% 정도 틀린다"처럼 비전문가도 쉽게 이해할 수 있습니다. 단, 실제값이 0에 가까우면 MAPE가 무한대로 커질 수 있어 주의가 필요합니다.

코드에서 중요한 부분은 inverse_transform입니다. 모델 학습 시 데이터를 정규화했기 때문에, 평가할 때는 원래 스케일로 복원해야 의미 있는 비교가 가능합니다.

김개발 씨는 여러 지표를 함께 계산하여 보고서를 다시 작성했습니다. "RMSE 150, MAE 120, MAPE 1.2%입니다.

평균적으로 1.2% 정도의 오차로 트래픽을 예측할 수 있습니다." 이제 팀장님도 이해할 수 있는 보고서가 완성되었습니다.

실전 팁

💡 - 비전문가에게 보고할 때는 MAPE처럼 직관적인 지표를 사용하세요

  • 지표는 항상 원본 스케일로 복원한 후 계산해야 의미가 있습니다
  • 베이스라인 모델과 비교하여 상대적인 개선 정도도 함께 보고하세요

7. 예측 API 서버 구축

모델이 완성되었지만 김개발 씨는 고민에 빠졌습니다. 이 모델을 다른 팀에서 어떻게 사용할 수 있을까요?

매번 주피터 노트북을 실행할 수는 없는 노릇입니다. 박시니어 씨가 해답을 제시합니다.

"API로 만들어서 배포하면 돼요. 누구나 HTTP 요청으로 예측 결과를 받을 수 있게요."

예측 API는 학습된 모델을 HTTP 서비스로 제공하는 것입니다. FastAPI를 사용하면 빠르고 안정적인 API 서버를 쉽게 구축할 수 있습니다.

다른 서비스에서 API를 호출하여 실시간으로 예측 결과를 받을 수 있습니다.

다음 코드를 살펴봅시다.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib

app = FastAPI(title="Traffic Prediction API")

# 모델과 전처리기 로드
model = joblib.load('traffic_model.pkl')
feature_extractor = joblib.load('feature_extractor.pkl')

class PredictionRequest(BaseModel):
    timestamp: str
    recent_traffic: list  # 최근 168시간의 트래픽 데이터

class PredictionResponse(BaseModel):
    predicted_traffic: float
    confidence_interval: dict

@app.post("/predict", response_model=PredictionResponse)
async def predict_traffic(request: PredictionRequest):
    try:
        features = feature_extractor.extract_single(
            request.timestamp, request.recent_traffic)
        prediction = model.predict([features])[0]
        return PredictionResponse(
            predicted_traffic=round(prediction, 2),
            confidence_interval={'lower': prediction*0.9, 'upper': prediction*1.1})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

모델을 완성한 김개발 씨에게 마케팅 팀에서 연락이 왔습니다. "트래픽 예측 결과를 우리 대시보드에 보여주고 싶은데, 어떻게 연동할 수 있을까요?" 운영 팀에서도 연락이 왔습니다.

"자동 스케일링 시스템에 예측 값을 반영하고 싶어요." 박시니어 씨가 말했습니다. "이제 모델을 API로 만들 차례예요.

다른 시스템에서 쉽게 사용할 수 있도록요." API란 무엇일까요? 쉽게 비유하자면, 이것은 마치 음식점 메뉴판과 같습니다.

손님은 주방에서 어떻게 요리하는지 몰라도 됩니다. 메뉴판에서 원하는 음식을 주문하면 주방에서 만들어서 가져다줍니다.

API도 마찬가지입니다. 다른 시스템은 모델이 어떻게 동작하는지 몰라도 됩니다.

그저 요청을 보내면 결과를 받을 수 있습니다. 위 코드에서 FastAPI를 사용하여 예측 서버를 구축합니다.

FastAPI는 파이썬 웹 프레임워크 중 가장 빠른 것 중 하나입니다. 또한 자동으로 API 문서를 생성해주어 매우 편리합니다.

PredictionRequest 클래스는 요청 형식을 정의합니다. timestamp는 예측하고 싶은 시점이고, recent_traffic은 최근 168시간(1주일)의 트래픽 데이터입니다.

이 데이터로 특성을 추출하여 예측에 사용합니다. PredictionResponse 클래스는 응답 형식을 정의합니다.

predicted_traffic은 예측된 트래픽 값이고, confidence_interval은 신뢰 구간입니다. 예측은 항상 불확실성을 가지므로 구간을 함께 제공하는 것이 좋습니다.

predict_traffic 함수가 실제 예측을 수행합니다. 먼저 feature_extractor로 특성을 추출하고, model.predict로 예측합니다.

에러가 발생하면 HTTPException으로 적절한 에러 응답을 반환합니다. 모델과 전처리기는 joblib으로 저장하고 로드합니다.

서버가 시작될 때 한 번만 로드하면 되므로 효율적입니다. 김개발 씨는 API를 배포한 후 마케팅 팀과 운영 팀에 문서를 공유했습니다.

이제 누구나 쉽게 트래픽 예측 기능을 사용할 수 있게 되었습니다.

실전 팁

💡 - Pydantic으로 요청/응답 형식을 명확히 정의하면 API 문서가 자동 생성됩니다

  • 모델 로드는 서버 시작 시 한 번만 하여 응답 속도를 높이세요
  • 예측 결과와 함께 신뢰 구간을 제공하면 더 유용합니다

8. 배치 예측 시스템 구현

API 서버가 잘 동작하고 있었지만, 운영 팀에서 새로운 요청이 왔습니다. "매일 아침 6시에 그날 24시간 동안의 트래픽을 미리 예측해서 알려줄 수 있나요?

스케일링 계획을 미리 세우고 싶어요." 김개발 씨는 배치 예측 시스템이 필요하다는 것을 깨달았습니다.

배치 예측은 정해진 시간에 대량의 예측을 한꺼번에 수행하는 것입니다. 실시간 API와 달리 미리 예측 결과를 생성하여 저장해둡니다.

스케줄러와 함께 사용하면 자동화된 예측 파이프라인을 구축할 수 있습니다.

다음 코드를 살펴봅시다.

from apscheduler.schedulers.blocking import BlockingScheduler
import pandas as pd
from datetime import datetime, timedelta

class BatchPredictor:
    def __init__(self, model, feature_extractor, db_conn):
        self.model = model
        self.feature_extractor = feature_extractor
        self.db = db_conn

    def predict_next_24h(self):
        # 최근 데이터 조회
        recent_data = self.db.query_recent_traffic(hours=168)
        predictions = []
        for hour_offset in range(24):
            target_time = datetime.now() + timedelta(hours=hour_offset)
            features = self.feature_extractor.extract_future(
                recent_data, target_time)
            pred = self.model.predict([features])[0]
            predictions.append({
                'timestamp': target_time, 'predicted_traffic': pred})
        # 예측 결과 저장
        self.db.save_predictions(predictions)
        print(f"[{datetime.now()}] 24시간 예측 완료: {len(predictions)}건")

# 스케줄러 설정
scheduler = BlockingScheduler()
scheduler.add_job(predictor.predict_next_24h, 'cron', hour=6, minute=0)

운영 팀의 요청은 합리적이었습니다. 매일 아침 그날의 예상 트래픽을 알 수 있다면 서버 자원을 미리 준비할 수 있습니다.

갑작스러운 트래픽 급증에 당황하는 일을 줄일 수 있습니다. 박시니어 씨가 설명했습니다.

"실시간 API와 배치 예측은 용도가 달라요. API는 지금 당장 필요한 예측을 위한 거고, 배치는 미리 준비하는 거예요." 배치 예측이란 무엇일까요?

쉽게 비유하자면, 이것은 마치 빵집에서 아침에 빵을 미리 굽는 것과 같습니다. 손님이 올 때마다 빵을 굽는 게 아니라, 아침 일찍 그날 팔릴 양을 예상하여 미리 구워둡니다.

덕분에 손님은 기다리지 않고 바로 빵을 살 수 있습니다. 위 코드의 BatchPredictor 클래스를 살펴보겠습니다.

predict_next_24h 메서드는 향후 24시간의 트래픽을 예측합니다. 먼저 query_recent_traffic으로 최근 168시간(1주일)의 데이터를 조회합니다.

이 데이터는 지연 특성을 계산하는 데 필요합니다. 그 다음 for 루프를 돌면서 각 시간대의 예측을 수행합니다.

extract_future 메서드는 미래 시점의 특성을 추출합니다. 현재 시점과 달리 미래에는 실제 트래픽 값이 없으므로 과거 데이터와 시간 특성만 활용합니다.

예측 결과는 리스트에 담긴 후 데이터베이스에 저장됩니다. 운영 팀은 이 테이블을 조회하여 스케일링 계획을 세울 수 있습니다.

APScheduler는 파이썬에서 가장 많이 사용되는 스케줄링 라이브러리입니다. 코드에서 cron 트리거를 사용하여 매일 오전 6시에 predict_next_24h를 실행하도록 설정했습니다.

마치 알람시계처럼 정해진 시간에 자동으로 작업이 실행됩니다. 김개발 씨는 배치 예측 시스템을 구축한 후 운영 팀에게 전달했습니다.

이제 매일 아침 자동으로 예측 결과가 생성되어 운영 효율성이 크게 높아졌습니다.

실전 팁

💡 - 배치 작업은 서비스 부하가 적은 새벽이나 이른 아침에 실행하세요

  • 배치 실행 로그를 남겨 장애 발생 시 원인을 파악할 수 있도록 하세요
  • 배치 실패 시 알림을 보내는 모니터링 시스템을 함께 구축하세요

9. 모델 모니터링 시스템

시스템이 배포된 지 한 달이 지났습니다. 그런데 어느 날 운영 팀에서 연락이 왔습니다.

"최근 예측이 자꾸 틀려요. 예측보다 실제 트래픽이 훨씬 높아요." 김개발 씨는 당황했습니다.

분명히 잘 동작하던 모델인데 왜 갑자기 정확도가 떨어진 걸까요?

모델 모니터링은 배포된 모델의 성능을 지속적으로 추적하는 것입니다. 시간이 지나면 데이터의 분포가 변하는 데이터 드리프트 현상이 발생할 수 있습니다.

모니터링을 통해 성능 저하를 조기에 발견하고 대응할 수 있습니다.

다음 코드를 살펴봅시다.

from scipy import stats
import logging

class ModelMonitor:
    def __init__(self, threshold_mape: float = 10.0):
        self.threshold = threshold_mape
        self.logger = logging.getLogger('model_monitor')

    def check_prediction_accuracy(self, predictions: pd.DataFrame,
                                   actuals: pd.DataFrame) -> dict:
        merged = predictions.merge(actuals, on='timestamp')
        mape = np.mean(np.abs(
            (merged['actual'] - merged['predicted']) / merged['actual'])) * 100
        status = 'OK' if mape < self.threshold else 'ALERT'
        if status == 'ALERT':
            self.logger.warning(f'Model accuracy degraded: MAPE={mape:.2f}%')
        return {'mape': mape, 'status': status}

    def detect_data_drift(self, training_data: pd.Series,
                          current_data: pd.Series) -> dict:
        # KS 검정으로 분포 변화 감지
        ks_stat, p_value = stats.ks_2samp(training_data, current_data)
        drift_detected = p_value < 0.05
        if drift_detected:
            self.logger.warning(f'Data drift detected: p-value={p_value:.4f}')
        return {'ks_statistic': ks_stat, 'p_value': p_value,
                'drift_detected': drift_detected}

박시니어 씨가 상황을 분석한 후 말했습니다. "아, 이거 데이터 드리프트 같네요.

지난달에 마케팅 캠페인이 있었잖아요. 사용자 패턴이 바뀐 거예요." 데이터 드리프트란 무엇일까요?

쉽게 비유하자면, 이것은 마치 계절이 바뀌는 것과 같습니다. 여름에 잘 팔리던 아이스크림이 겨울에는 잘 안 팔립니다.

세상이 변하면 과거의 패턴도 변합니다. 머신러닝 모델도 마찬가지입니다.

학습 시점의 데이터와 현재 데이터가 다르면 예측 성능이 떨어집니다. 위 코드의 ModelMonitor 클래스는 두 가지를 감시합니다.

첫 번째는 예측 정확도 모니터링입니다. check_prediction_accuracy 메서드는 예측값과 실제값을 비교하여 MAPE를 계산합니다.

MAPE가 임계값(여기서는 10%)을 넘으면 ALERT 상태가 되어 경고 로그를 남깁니다. 두 번째는 데이터 드리프트 감지입니다.

detect_data_drift 메서드는 KS 검정(Kolmogorov-Smirnov test)을 사용합니다. 이 검정은 두 데이터 집합의 분포가 같은지 확인합니다.

p-value가 0.05 미만이면 분포가 유의미하게 다르다고 판단합니다. 왜 이런 모니터링이 중요할까요?

모델은 학습 시점의 데이터로만 훈련됩니다. 하지만 세상은 계속 변합니다.

신규 서비스가 출시되거나, 마케팅 캠페인이 진행되거나, 계절이 바뀌면 사용자 행동 패턴도 변합니다. 모니터링 없이는 모델이 언제부터 잘못된 예측을 하는지 알 수 없습니다.

김개발 씨는 모니터링 시스템을 구축한 후 매일 자동으로 성능을 체크하도록 설정했습니다. 데이터 드리프트가 감지되면 Slack으로 알림이 오도록 연동했습니다.

이제 문제가 발생하면 바로 대응할 수 있게 되었습니다.

실전 팁

💡 - 예측 정확도 모니터링은 최소 하루 단위로 수행하세요

  • 데이터 드리프트 임계값은 도메인에 맞게 조정하세요
  • 알림이 너무 자주 오면 무시하게 되니 적절한 임계값을 설정하세요

10. 자동 재학습 파이프라인

모니터링 시스템이 데이터 드리프트를 감지했습니다. 김개발 씨는 새 데이터로 모델을 다시 학습시켜야 합니다.

하지만 매번 수동으로 재학습을 진행하려니 너무 번거로웠습니다. 박시니어 씨가 제안합니다.

"자동 재학습 파이프라인을 만들어요. 드리프트가 감지되면 자동으로 새 모델을 학습시키는 거예요."

자동 재학습 파이프라인은 모델 성능 저하 시 새로운 데이터로 자동으로 모델을 재학습시킵니다. 이를 MLOps의 핵심 요소 중 하나입니다.

사람의 개입 없이도 모델이 최신 상태를 유지할 수 있습니다.

다음 코드를 살펴봅시다.

import mlflow
from datetime import datetime

class AutoRetrainer:
    def __init__(self, model_class, data_pipeline, monitor):
        self.model_class = model_class
        self.data_pipeline = data_pipeline
        self.monitor = monitor

    def run_retraining(self) -> str:
        with mlflow.start_run():
            # 최신 데이터 수집 및 전처리
            recent_data = self.data_pipeline.collect_recent(months=3)
            processed_data = self.data_pipeline.preprocess(recent_data)
            features = self.data_pipeline.extract_features(processed_data)
            # 새 모델 학습
            new_model = self.model_class()
            new_model.train(features)
            metrics = new_model.evaluate()
            # MLflow에 메트릭 기록
            mlflow.log_metrics(metrics)
            mlflow.sklearn.log_model(new_model, 'traffic_model')
            # 새 모델이 더 좋으면 배포
            if metrics['mape'] < self.monitor.threshold:
                new_model.deploy()
                return f"New model deployed: MAPE={metrics['mape']:.2f}%"
            return f"Retraining completed but not deployed: MAPE={metrics['mape']:.2f}%"

김개발 씨는 지난 한 달간 수동으로 세 번이나 모델을 재학습시켰습니다. 데이터를 다운로드하고, 전처리하고, 학습시키고, 평가하고, 배포하는 과정이 반복되었습니다.

매번 같은 과정을 반복하다 보니 실수도 생기고 시간도 많이 걸렸습니다. 박시니어 씨가 말했습니다.

"반복되는 작업은 자동화해야 해요. 이게 바로 MLOps예요." MLOps란 무엇일까요?

쉽게 비유하자면, 이것은 마치 자동차 공장의 생산 라인과 같습니다. 사람이 일일이 부품을 조립하는 대신, 로봇이 자동으로 조립합니다.

품질 검사도 자동으로 이루어지고, 문제가 있으면 생산 라인이 멈춥니다. MLOps도 마찬가지로 모델의 학습, 평가, 배포를 자동화합니다.

위 코드의 AutoRetrainer 클래스를 살펴보겠습니다. run_retraining 메서드는 전체 재학습 과정을 자동화합니다.

먼저 최근 3개월의 데이터를 수집합니다. 너무 오래된 데이터는 현재 패턴을 반영하지 못하므로 적절한 기간을 선택하는 것이 중요합니다.

수집한 데이터는 전처리와 특성 추출 과정을 거칩니다. 새 모델을 학습시킨 후에는 반드시 평가를 수행합니다.

새 모델이 기존 모델보다 나쁠 수도 있기 때문입니다. MAPE가 임계값보다 낮을 때만 새 모델을 배포합니다.

MLflow는 머신러닝 실험을 추적하고 관리하는 오픈소스 플랫폼입니다. mlflow.start_run()으로 실험을 시작하고, log_metrics로 성능 지표를 기록하고, log_model로 모델을 저장합니다.

나중에 어떤 데이터로 어떤 모델을 학습시켰는지 추적할 수 있습니다. 김개발 씨는 자동 재학습 파이프라인을 구축한 후 모니터링 시스템과 연동했습니다.

이제 데이터 드리프트가 감지되면 자동으로 재학습이 시작됩니다. 사람의 개입 없이도 모델이 항상 최신 상태를 유지합니다.

실전 팁

💡 - 재학습 전후의 성능을 반드시 비교하여 새 모델이 더 좋을 때만 배포하세요

  • MLflow, Weights&Biases 등의 실험 추적 도구를 활용하세요
  • 재학습 주기와 데이터 기간은 도메인에 맞게 조정하세요

11. 종합 테스트 및 배포

드디어 모든 구성 요소가 완성되었습니다. 데이터 수집, 전처리, 모델 학습, API 서버, 배치 예측, 모니터링, 자동 재학습까지.

하지만 박시니어 씨가 말했습니다. "아직 끝이 아니에요.

전체 시스템이 제대로 동작하는지 테스트해야 해요. 그리고 프로덕션에 안전하게 배포하는 방법도 고려해야 하고요."

종합 테스트는 개별 컴포넌트가 아닌 전체 시스템의 동작을 검증합니다. 배포 전략은 새 버전을 안전하게 프로덕션에 적용하는 방법입니다.

카나리 배포나 블루-그린 배포를 사용하면 위험을 최소화할 수 있습니다.

다음 코드를 살펴봅시다.

import pytest
from unittest.mock import Mock

class IntegrationTest:
    def setup_method(self):
        self.system = TrafficPredictionSystem('config/test.yaml')

    def test_full_pipeline(self):
        # 전체 파이프라인 테스트
        result = self.system.run_pipeline()
        assert 'rmse' in result
        assert result['rmse'] < 200  # 허용 임계값

    def test_api_endpoint(self, client):
        # API 엔드포인트 테스트
        response = client.post('/predict', json={
            'timestamp': '2024-01-15T10:00:00',
            'recent_traffic': [100]*168
        })
        assert response.status_code == 200
        assert 'predicted_traffic' in response.json()

# 배포 스크립트
def deploy_with_canary(new_version: str, canary_percent: int = 10):
    # 10%의 트래픽만 새 버전으로 라우팅
    load_balancer.set_canary_weight(new_version, canary_percent)
    # 1시간 동안 모니터링
    monitor.observe(duration_hours=1)
    if monitor.is_healthy():
        load_balancer.set_weight(new_version, 100)  # 전체 전환
        print(f"Deployment successful: {new_version}")
    else:
        load_balancer.rollback()
        raise DeploymentError("Canary deployment failed")

김개발 씨는 모든 코드를 작성했습니다. 단위 테스트도 통과했습니다.

하지만 박시니어 씨가 말했습니다. "개별 부품이 잘 동작한다고 전체 시스템이 잘 동작하는 건 아니에요.

레고 블록 하나하나가 멀쩡해도 조립이 잘못되면 무너지잖아요." 종합 테스트(통합 테스트)는 왜 필요할까요? 쉽게 비유하자면, 이것은 마치 자동차 조립 후 시운전과 같습니다.

엔진, 바퀴, 핸들이 각각 잘 작동해도 조립 후에 제대로 달리는지 확인해야 합니다. 시스템도 마찬가지입니다.

데이터 수집, 전처리, 모델, API가 올바르게 연결되어 있는지 전체 흐름을 테스트해야 합니다. 위 코드에서 IntegrationTest 클래스는 전체 파이프라인을 테스트합니다.

test_full_pipeline 메서드는 데이터 수집부터 모델 평가까지 전체 과정을 실행하고 결과를 검증합니다. test_api_endpoint 메서드는 API가 올바른 응답을 반환하는지 확인합니다.

카나리 배포란 무엇일까요? 광산에서 카나리아(새)를 데리고 들어가면 유독 가스를 먼저 감지할 수 있습니다.

소프트웨어 배포도 마찬가지입니다. 새 버전을 전체 사용자에게 바로 적용하지 않고, 일부(예: 10%)에게만 먼저 적용합니다.

문제가 없으면 점차 비율을 높이고, 문제가 발생하면 즉시 이전 버전으로 롤백합니다. deploy_with_canary 함수는 카나리 배포를 구현합니다.

처음에는 10%의 트래픽만 새 버전으로 보냅니다. 1시간 동안 모니터링하여 문제가 없으면 100%로 전환합니다.

문제가 발생하면 자동으로 롤백합니다. 김개발 씨는 종합 테스트를 모두 통과한 후 카나리 배포를 진행했습니다.

처음에는 긴장되었지만, 시스템이 안정적으로 동작하는 것을 확인하고 안도했습니다. 이제 트래픽 예측 시스템이 프로덕션에서 완벽하게 동작합니다.

박시니어 씨가 말했습니다. "축하해요, 김개발 씨.

이제 당신도 데이터 시스템을 처음부터 끝까지 구축할 수 있는 개발자가 되었네요."

실전 팁

💡 - 프로덕션과 동일한 환경에서 종합 테스트를 수행하세요

  • 카나리 배포 시 핵심 지표(응답 시간, 에러율)를 면밀히 모니터링하세요
  • 롤백 계획을 항상 준비하고, 롤백 연습도 주기적으로 수행하세요

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

#Python#TrafficPrediction#MachineLearning#TimeSeriesAnalysis#DataPipeline#Data Science

댓글 (0)

댓글을 작성하려면 로그인이 필요합니다.