4.5HZ - AI 기반 음악 추천 서비스 개발기

4.5HZ - AI 기반 음악 추천 서비스 개발기

프로젝트 기간: 2023.11.27 ~ 2024.01.05 (40일)
팀 구성: 5명 (이도형, 유현준, 정유진, 조민진, 최세인)
역할: 백엔드 개발, 데이터 파이프라인 구축, ML 모델 개발

음악 데이터 수집부터 추천까지 연결한 서비스

이 프로젝트는 글로벌 음악 데이터를 분석해 사용자에게 음악을 추천하는 웹 서비스입니다. Spotify API로 데이터를 수집하고 Apache Airflow로 처리 작업을 자동화했으며, 추천 모델의 결과를 웹 서비스에서 제공하는 흐름까지 구현했습니다.

프로젝트 결과

• 최종 프로젝트 최우수 수상
• 추천 정확도: RMSE 0.85 달성
• 데이터 처리량: 100만+ 음악 트랙 분석
• 실시간 처리: 1시간 간격 데이터 업데이트

수집부터 추천까지 이어지는 시스템 구조

전체 처리 흐름

구성 요소와 데이터 흐름

graph TB
    %% 데이터 수집 및 전처리
    A[Spotify API] --> B[pandas<br/>Data Preprocessing]
    C[Dummy Data] --> B
    D[Apache Airflow<br/>Orchestration] --> B
    
    %% 데이터 저장
    B --> E[MySQL<br/>Main Database]
    B --> F[OpenSearch<br/>Log Search]
    
    %% 실시간 스트리밍 파이프라인
    G[User Interaction] --> H[fluentd<br/>Log Collector]
    H --> I[Kafka<br/>Message Queue]
    I --> J[Spark Streaming<br/>Real-time Processing]
    J --> K[MongoDB<br/>Log Storage]
    
    %% 시각화 및 분석
    K --> L[Chart.js<br/>Visualization]
    L --> M[Apache Airflow<br/>Analytics Pipeline]
    M --> E
    
    %% 웹 서비스 및 추천
    E --> N[Django<br/>Web Service]
    N --> O[Content/Collaborative<br/>Recommendation Algorithm]
    O --> N
    O --> M
    
    %% 최종 서비스
    N --> P[4.5HZ<br/>Music Recommendation Service]
    
    %% 스타일링
    classDef api fill:#1db954,stroke:#1ed760,color:#fff
    classDef processing fill:#ff6b35,stroke:#ff8c42,color:#fff
    classDef storage fill:#336791,stroke:#4a90a4,color:#fff
    classDef streaming fill:#ff6b6b,stroke:#ff8e8e,color:#fff
    classDef web fill:#092e20,stroke:#0c4b33,color:#fff
    classDef service fill:#6c5ce7,stroke:#a29bfe,color:#fff
    
    class A api
    class B,D,M processing
    class E,F,K storage
    class G,H,I,J streaming
    class N,O web
    class P service

데이터베이스 구조

구현에 사용한 기술

백엔드와 웹 프레임워크

• Django 4.2.8: 웹 프레임워크
• Django REST Framework: API 개발
• MySQL: 메인 데이터베이스
• MongoDB: 로그 데이터 저장

데이터 처리

• Apache Spark 3.2.4: 대용량 데이터 처리
• Apache Kafka: 실시간 스트리밍
• Apache Airflow: 작업 일정과 실행 순서 관리
• Fluentd: 로그 수집

머신러닝

• Scikit-learn: 머신러닝 알고리즘
• TensorFlow: 딥러닝 모델
• Surprise: 추천 시스템 라이브러리
• Pandas, NumPy: 데이터 처리

음악 데이터 수집과 처리

1. 데이터 수집

Spotify API에서 장르별 아티스트를 찾고, 각 아티스트의 인기 곡과 오디오 특성을 가져왔습니다. 수집한 특성은 이후 전처리와 콘텐츠 기반 추천에 사용했습니다.

# Spotify API를 통한 음악 데이터 수집
import spotipy
from spotipy.oauth2 import SpotifyClientCredentials

def collect_track_data():
    sp = spotipy.Spotify(client_credentials_manager=SpotifyClientCredentials())
    
    # 아티스트 정보 수집
    artists = sp.search(q='genre:k-pop', type='artist', limit=50)
    
    # 트랙 정보 및 오디오 특성 수집
    for artist in artists['artists']['items']:
        tracks = sp.artist_top_tracks(artist['id'])
        for track in tracks['tracks']:
            audio_features = sp.audio_features(track['id'])
            # 데이터베이스 저장

2. 데이터 전처리

곡마다 값의 범위가 다른 오디오 특성을 표준화한 뒤 K-means로 묶었습니다. 아래 코드는 춤추기 적합한 정도, 에너지, 음악의 정서적 성격, 템포를 입력으로 사용합니다.

# 오디오 특성을 활용한 클러스터링
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

def preprocess_audio_features(df):
    # 오디오 특성 정규화
    features = ['danceability', 'energy', 'valence', 'tempo']
    scaler = StandardScaler()
    scaled_features = scaler.fit_transform(df[features])
    
    # K-means 클러스터링
    kmeans = KMeans(n_clusters=10, random_state=42)
    df['cluster'] = kmeans.fit_predict(scaled_features)
    
    return df

3. 실시간 데이터 처리

사용자 로그는 Kafka에서 받아 Spark Streaming으로 집계했습니다. 사용자별 재생 횟수를 시간 구간별로 계산하는 부분은 다음과 같습니다.

# Kafka + Spark Streaming을 통한 실시간 로그 처리
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

def process_user_logs():
    spark = SparkSession.builder.appName("UserLogProcessor").getOrCreate()
    
    # Kafka에서 로그 데이터 읽기
    df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "localhost:9092") \
        .option("subscribe", "user_logs") \
        .load()
    
    # 실시간 집계
    result = df.groupBy("user_id", window("timestamp", "1 hour")) \
               .agg(count("track_id").alias("play_count"))
    
    return result

협업 필터링과 콘텐츠 기반 추천을 결합했다

1. 협업 필터링

사용자별 평점으로 SVD 모델을 학습하고, 아직 듣지 않은 곡의 예상 평점을 계산했습니다. 추천 목록은 예상 평점이 높은 순서로 구성했습니다.

from surprise import SVD, Dataset, Reader
from surprise.model_selection import cross_validate

def build_collaborative_model():
    # 사용자-아이템 매트릭스 구성
    reader = Reader(rating_scale=(0, 5))
    data = Dataset.load_from_df(ratings_df, reader)
    
    # SVD 모델 학습
    model = SVD(n_epochs=20, lr_all=0.005, reg_all=0.4)
    cross_validate(model, data, measures=['RMSE', 'MAE'], cv=5)
    
    return model

def recommend_songs(user_id, n_recommendations=10):
    # 사용자가 아직 듣지 않은 곡들 중에서 추천
    user_tracks = get_user_played_tracks(user_id)
    all_tracks = get_all_tracks()
    unplayed_tracks = set(all_tracks) - set(user_tracks)
    
    predictions = []
    for track_id in unplayed_tracks:
        pred = model.predict(user_id, track_id)
        predictions.append((track_id, pred.est))
    
    # 예상 평점이 높은 순으로 정렬
    predictions.sort(key=lambda x: x[1], reverse=True)
    return predictions[:n_recommendations]

2. 콘텐츠 기반 필터링

기준 곡과 다른 곡의 오디오 특성을 코사인 유사도로 비교했습니다. 청취 이력을 이용하는 협업 필터링과 함께, 곡 자체의 특성으로도 추천 후보를 만들었습니다.

from sklearn.metrics.pairwise import cosine_similarity

def content_based_recommendation(track_id, n_recommendations=10):
    # 오디오 특성 기반 유사도 계산
    track_features = get_track_audio_features(track_id)
    all_features = get_all_track_features()
    
    # 코사인 유사도 계산
    similarities = cosine_similarity([track_features], all_features)[0]
    
    # 유사도가 높은 곡들 반환
    similar_indices = similarities.argsort()[-n_recommendations-1:-1][::-1]
    return [get_track_by_index(idx) for idx in similar_indices]

3. 하이브리드 추천 시스템

협업 필터링과 콘텐츠 기반 추천 결과를 가중 평균으로 결합했습니다. 기준 곡이 없으면 사용자가 최근에 재생한 곡들로 콘텐츠 기반 추천 후보를 모았습니다.

def hybrid_recommendation(user_id, track_id=None):
    # 협업 필터링 결과
    cf_recommendations = collaborative_filtering(user_id)
    
    # 콘텐츠 기반 필터링 결과
    if track_id:
        cb_recommendations = content_based_recommendation(track_id)
    else:
        # 사용자의 최근 재생 곡 기반
        recent_tracks = get_user_recent_tracks(user_id, limit=5)
        cb_recommendations = []
        for track in recent_tracks:
            cb_recommendations.extend(content_based_recommendation(track['id']))
    
    # 가중 평균으로 최종 추천 리스트 생성
    final_recommendations = weighted_combine(cf_recommendations, cb_recommendations)
    return final_recommendations

Airflow로 처리 순서를 자동화했다

Apache Airflow DAG

전처리, 모델 재학습, 집계 데이터 생성, 데이터베이스 갱신 순으로 작업을 연결했습니다. 앞 단계가 끝나면 다음 단계가 실행되도록 DAG에 의존 관계를 지정했습니다.

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

default_args = {
    'owner': '4.5HZ',
    'depends_on_past': False,
    'start_date': datetime(2023, 12, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'music_recommendation_pipeline',
    default_args=default_args,
    description='음악 추천 시스템 파이프라인',
    schedule_interval='@hourly',
    catchup=False,
    tags=['music', 'recommendation', 'ml'],
)

# 1. 데이터 전처리
preprocess_task = SparkSubmitOperator(
    task_id='preprocess_data',
    application='/home/ubuntu/airflow/dags/preprocess.py',
    conn_id='spark_local',
    dag=dag,
)

# 2. 모델 재학습
retrain_model = PythonOperator(
    task_id='retrain_model',
    python_callable=retrain_recommendation_model,
    dag=dag,
)

# 3. 집계 데이터 생성
aggregate_data = SparkSubmitOperator(
    task_id='aggregate_data',
    application='/home/ubuntu/airflow/dags/aggregate.py',
    conn_id='spark_local',
    dag=dag,
)

# 4. 데이터베이스 업데이트
update_database = PythonOperator(
    task_id='update_database',
    python_callable=update_recommendation_database,
    dag=dag,
)

# 태스크 의존성 설정
preprocess_task >> retrain_model >> aggregate_data >> update_database

조회 성능과 추천 응답 시간 개선

1. 데이터베이스 최적화

조회에 사용하는 필드에 인덱스를 설정하고, 장르별 인기 곡을 가져오는 쿼리를 작성했습니다.

# 인덱스 최적화
class Track(models.Model):
    track_id = models.CharField(primary_key=True, max_length=30)
    track_name = models.CharField(max_length=200, db_index=True)
    genres = models.CharField(max_length=80, db_index=True)
    
    class Meta:
        indexes = [
            models.Index(fields=['track_popularity']),
            models.Index(fields=['release_date']),
        ]

# 쿼리 최적화
def get_popular_tracks_by_genre(genre, limit=50):
    return Track.objects.filter(
        genres__icontains=genre
    ).order_by('-track_popularity')[:limit]

2. 캐싱 전략

사용자별 추천 결과를 캐시에 저장했습니다. 캐시에 결과가 있으면 바로 반환하고, 없을 때 추천 알고리즘을 실행한 뒤 결과를 저장합니다.

from django.core.cache import cache
import json

def get_user_recommendations(user_id):
    cache_key = f"recommendations_{user_id}"
    cached_result = cache.get(cache_key)
    
    if cached_result:
        return json.loads(cached_result)
    
    # 추천 알고리즘 실행
    recommendations = hybrid_recommendation(user_id)
    
    # 1시간 캐싱
    cache.set(cache_key, json.dumps(recommendations), 3600)
    return recommendations

컨테이너 배포와 추천 로그 기록

Docker 컨테이너화

Python 실행 환경과 의존성을 컨테이너에 담고 Gunicorn으로 Django 애플리케이션을 실행하도록 구성했습니다.

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "--bind", "0.0.0.0:8000", "final_pjt.wsgi:application"]

로깅 및 모니터링

추천 요청마다 사용자 식별자, 추천 목록, 응답 시간, 실행 시각을 JSON 형식의 로그로 남겼습니다.

import logging
import json
from datetime import datetime

class RecommendationLogger:
    def __init__(self):
        self.logger = logging.getLogger('recommendation')
        handler = logging.FileHandler('logs/recommendation.log')
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
    
    def log_recommendation(self, user_id, recommendations, response_time):
        log_data = {
            'user_id': user_id,
            'recommendations': recommendations,
            'response_time': response_time,
            'timestamp': datetime.now().isoformat()
        }
        self.logger.info(json.dumps(log_data))

시연과 성능 지표

시연 영상

• YouTube 시연 영상: 4.5HZ 음악 추천 서비스 시연
• 로컬 시연 영상: 4.5HZ 홈페이지 시연 영상

성능 지표

• 추천 정확도: RMSE 0.85 (Surprise 라이브러리 기준)
• 응답 시간: 평균 200ms 이하
• 시스템 가용성: 99.5%
• 데이터 처리량: 시간당 10만+ 로그 처리

사용자 피드백

• 추천 만족도: 4.2/5.0
• 재방문율: 65%
• 평균 세션 시간: 12분

프로젝트에서 익힌 것

기술 구현

1. 풀스택 개발: 프론트엔드부터 데이터 파이프라인까지 전체 시스템 구축
2. MLOps: 모델 학습부터 배포까지의 자동화된 파이프라인 구축
3. 대용량 데이터 처리: Spark를 활용한 분산 처리 경험
4. 실시간 시스템: Kafka + Spark Streaming을 통한 실시간 데이터 처리

팀 개발

1. 팀 협업: 5명으로 구성된 팀의 협업 방식 정리
2. 코드 리뷰: Git을 활용한 체계적인 코드 리뷰 프로세스
3. 문서화: 기술 문서 작성 및 API 문서화 경험

참고 자료

• Spotify Web API Documentation
• Apache Spark Documentation
• Django Documentation
• Surprise Library Documentation

---

회고

4.5HZ에서는 데이터 수집과 전처리, 모델 학습, 웹 서비스 연결, 배포와 로그 처리를 하나의 서비스 흐름으로 구성했습니다.

추천 성능을 확인하는 일과 함께, 데이터를 계속 갱신하고 결과를 서비스에 반영하는 구조도 다뤘습니다. 이 과정에서 데이터 처리 흐름과 MLOps가 추천 모델만큼 중요하다는 점을 확인했습니다.

---

관련 링크

GitHub 저장소

• 소스 코드 및 문서: MULTI_PJT2_4.5HZ

YouTube 시연 영상

• 실제 서비스 데모: 4.5HZ 음악 추천 서비스 시연

발표 자료

• 프로젝트 발표 PDF: 4.5HZ 발표자료