[PySpark] 컨텐츠 기반 영화 추천 시스템 만들어보기

🔊 본 포스팅은 PySpark를 활용한 Kaggle Notebook을 필사하면서 배우게 된 코드 내용을 기반으로 포스팅했음을 알려드립니다. 또한 앞으로 소개될 PySpark의 문법에 대해서 생소하시다면 여기를 참고해 간단한 예시를 통해 이해를 할 수 있습니다.

 

이번 포스팅에서는 PySpark 그 중에서도 Spark SQL 과 Spark MLlib을 활용해 잠재요인 기반의 영화 추천 시스템을 구현해보려 한다. 활용한 데이터는 Kaggle의 무비렌즈 데이터의 movies, ratings 데이터를 필자가 직접 전처리한 데이터를 활용했다. 전처리된 데이터를 활용하려면 여기를 클릭해 다운로드 받자.

 

영화를 추천해주는 시스템을 만들어보자

 

추천 시스템 중에는 컨텐츠 기반 추천, 협업 기반 추천 등의 여러 종류가 있지만 여기서는 다른 사용자들의 제품에 대한 평점을 기반으로 하여 다른 제품을 추천해주는 협업 필터링 방법의 잠재적 요인 기반 모델을 사용하려고 한다.

1. 필요한 라이브러리와 데이터 로드

필요한 라이브러리들을 임포트하고 PySpark를 활용하기 위한 스파크 세션을 만들어보자. 그리고 로드한 데이터를 Pandas 형태로 미리보기해서 어떤 피처들이 존재하는지 살펴보자.

 

import sklearn
import random 

# Pyspark Library #
# SQL
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext
from pyspark.sql.functions import mean, col, split, regexp_extract, when, lit
# ML
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler, IndexToString
from pyspark.ml.feature import QuantileDiscretizer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

import os
import pandas as pd
import numpy as np
os.chdir('/Users/younghun/Desktop/gitrepo/data/movie_dataset/')

# 스파크 세션 만들기
spark = SparkSession\
        .builder\
        .appName('recommender_system')\
        .getOrCreate()

df = spark.read.csv(os.getcwd() + '/movie_ratings_df.csv',
                   inferSchema=True, header=True)
df.limit(3).toPandas()

 

데이터 로드 화면

 

위 화면에서 보는 것처럼 유저를 나타내는 userId, 영화의 제목인 title, 그 해당 영화에 대한 평점인 rating 변수들이 존재한다.

2. 데이터 전처리

문자열 형태로 이루어져있는 영화 제목의 값들을 수치형으로 바꾸어주자.

 

from pyspark.ml.feature import StringIndexer, VectorAssembler, IndexToString

stringIndexer = StringIndexer(inputCol='title',
                             outputCol='title_new')
model = stringIndexer.fit(df)
indexed = model.transform(df)
indexed.limit(5).toPandas()

 

영화 제목을 숫자의 값으로 인코딩 하자

3. 잠재 요인 기반 추천 시스템 ALS 모델 만들고 평가해보기

구축할 추천 시스템은 잠재 요인을 사용해 만드는 그 중에서도 SGD(Stochastic Gradient Descent)가 아닌 ALS(Alternate Least Squares) 모델을 사용해보려 한다. 편리하게도 PySpark에서도 ALS 모델을 쉽게 구축할 수 있는 라이브러리를 제공한다. ALS 메소드의 주요 파라미터를 잠깐 살펴보고 코드로 구현해보자.

 

• regParam : 데이터셋 크기에 의존하지 않고 추천 시스템의 일반화를 시켜주기위한 정규화 항이다.
• coldStartStrategy = 'drop' : 아직 평가되지 않은 즉, 결측치 값을 갖는 데이터를 제외하고 모델의 성능이 평가된다.(참고로 'nan' 값으로 할당하면 결측치를 포함해 모델 성능을 평가한다.)
• nonnegative = True : Least Squares를 할 때 음수값이 미포함되어 있는 제한사항을 줄 것인지를 선택한다. 이번에 활용된 영화 데이터의 평점과 영화제목, 유저 아이디는 모두 음수값이 없기 때문에 True로 하는 것이 바람직하다. 이 NNF(Non-Negative Factorization 개념에 대해서는 여기를 참고해보자)

참고로 Cold Start란, 새로운 유저가 나타났을 때, 새로운 유저에 대한 히스토리가 없기 때문에 이 유저에게 제품울 추천해줄 수 없는 현상을 말한다.

 

# ALS recommender algorithm
from pyspark.ml.recommendation import ALS

train, test = indexed.randomSplit([0.75, 0.25])

rec = ALS(maxIter=10,
         regParam=0.01,
         userCol='userId',
         itemCol='title_new',
         ratingCol='rating', # label -> predict할 때는 필요 없음!
         nonnegative=True,
         coldStartStrategy='drop')
# ALS모델 학습 -> dataframe을 넣어주기
rec_model = rec.fit(train)

# transform을 이용해 예측 -> dataframe을 넣어주기
pred_ratings = rec_model.transform(test)
pred_ratings.limit(5).toPandas()

 

ALS 모델로 예측한 결과

 

이제 예측값을 계산했으니 실제 평점과 예측 평점 사이의 차이값으로 RMSE와 MAE값을 측정해보자.

 

# Get metric for training
from pyspark.ml.evaluation import RegressionEvaluator

evaluator = RegressionEvaluator(labelCol='rating',
                               predictionCol='prediction',
                               metricName='rmse')
# evaluate 메소드에 예측값 담겨있는 dataframe 넣어주기
rmse = evaluator.evaluate(pred_ratings)

mae_eval = RegressionEvaluator(labelCol='rating',
                              predictionCol='prediction',
                              metricName='mae')
mae = mae_eval.evaluate(pred_ratings)

print("RMSE:", rmse)
print("MAE:", mae)

 

평가 결과

4. 특정 유저가 좋아할만한 영화를 추천해주기

이제 특정 유저를 파라미터로 입력시켰을 때 ALS 추천 모델을 적용해 그 특정 유저가 좋아할만한 영화를 자동으로 추천해주는 함수를 만들어보자. 함수를 정의할 때 필자가 주석으로 자세하게 코드 한줄한줄 설명을 써놓았다.

 

import sklearn
import random
import os
import pandas as pd
import numpy as np
os.chdir('/Users/younghun/Desktop/gitrepo/data/movie_dataset/')

# Pyspark Library #
# SQL
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext
from pyspark.sql.functions import mean, col, split, regexp_extract, when, lit
# ML
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler, IndexToString
from pyspark.ml.feature import QuantileDiscretizer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.recommendation import ALS

df = spark.read.csv(os.getcwd() + '/movie_ratings_df.csv',
                   inferSchema=True, header=True)

stringIndexer = StringIndexer(inputCol='title',
                             outputCol='title_new')
model = stringIndexer.fit(df)
indexed = model.transform(df)

# 숫자로 바꾼 영화제목들 중 Unique한 값들만 담아 추출하기 -> Dataframe 반환
unique_movies = indexed.select("title_new").distinct()

def top_movies(user_id, n):
    """
    특정 user_id가 좋아할 만한 n개의 영화 추천해주는 함수
    """
    # unique_movies 데이터프레임을 'a'라는 데이터프레임으로 alias시키기
    a = unique_movies.alias('a')

    # 특정 user_id가 본 영화들만 담은 새로운 데이터프레임 생성
    watched_movies = indexed.filter(indexed['userId'] == user_id)\
                            .select('title_new')

    # 특정 user_id가 본 영화들을 'b'라는 데이터프레임으로 alias시키기
    b = watched_movies.alias('b')

    # unique_movies를 기준으로 watched_movies를 조인시켜서 user_id가 보지 못한 영화들 파악 가능
    total_movies = a.join(b, a['title_new'] == b['title_new'],
                         how='left')

    # b 데이터프레임의 title_new값이 결측치를 갖고 있는 행의 a.title_new를 뽑아냄으로써 user_id가 아직 못본 영화들 추출
    # col('b.title_new') => b 데이터프레임의 title_new칼럼 의미(SQL처럼 가능!)
    remaining_movies = total_movies\
                       .where(col('b.title_new').isNull())\
                       .select('a.title_new').distinct()
    # remaining_movies 데이터프레임에 특정 user_id값을 동일하게 새로운 변수로 추가해주기
    remaining_movies = remaining_movies.withColumn('userId',
                                                  lit(int(user_id)))
    # 위에서 만든 ALS 모델을 사용하여 추천 평점 예측 후 n개 만큼 view -> 
    recommender = rec_model.transform(remaining_movies)\
                           .orderBy('prediction', ascending=False)\
                           .limit(n)
    # StringIndexer로 만든 것을 역으로 바꾸기 위해 IndexToString사용(영화제목을 숫자->한글제목)
    movie_title = IndexToString(inputCol='title_new',
                               outputCol='title',
                               labels=model.labels) #여기서 model.labels는 StringIndexer에서 fit시켰을 때 생긴 레이블. 즉, 영화 제목들
    # transform해서 영화제목을 숫자->한글로 변환! => dataframe으로 반환
    final_recommendations = movie_title.transform(recommender)

    return final_recommendations.show(n, truncate=False)

# userid가 1817번인 유저가 볼만한 영화 상위 5개 추천해주기
top_movies(1817, 5)

 

1817번 유저가 볼만한 5개의 영화