01.3_오디오스크로블러_데이터셋으로_음악추천하기.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. 데이터셋\n",
    "- 오디오스크로블러 (Audioscrobbler) 2005년 공개 데이터셋\n",
    "- http://www.iro.umontreal.ca/~lisa/datasets/profiledata_06-May-2005.tar.gz\n",
    "- user_artist_data.txt : 141,000명 사용자, 160만 아티스트 정보, 2,420만 건 음악 재생정보 (with 재생횟수)\n",
    "- artist_data.txt : 각각의 아티스트에 ID 부여됨, 잘못 기록된 아티스트 ID는 대표 ID와 연결됨"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from urllib.request import urlretrieve\n",
    "\n",
    "f = urlretrieve(\"http://www.iro.umontreal.ca/~lisa/datasets/profiledata_06-May-2005.tar.gz\", \n",
    "                       \"profiledata_06-May-2005.tar.gz\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. 교차 최소 제곱 추천 알고리즘\n",
    "- 협업 필터링, collaborative filtering\n",
    "    - https://en.wikipedia.org/wiki/Collaborative_filtering\n",
    "- 잠재 요인 모델, Factor analysis\n",
    "    - https://en.wikipedia.org/wiki/Factor_analysis\n",
    "- 행렬 분해\n",
    "    - https://en.wikipedia.org/wiki/Non-negative_matrix_factorization\n",
    "- 교차 최소 제곱 알고리즘, Alternating Least Squares Method for Collaborative Filtering (ALS)\n",
    "    - http://bugra.github.io/work/notes/2014-04-19/alternating-least-squares-method-for-collaborative-filtering/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## (1) 행렬 분해\n",
    "\n",
    "- A : 사용자 i가 아티스트 j의 음악을 들었다면 A의 i행 j열에 값이 존재하는 행렬\n",
    "- 사용자 - 아티스트의 가능한 모든 조합 중 오직 극소수만이 실제 데이터로 등장하기 때문에 원소의 대부분의 값들은 0임 (sparse matrix)\n",
    "- 행렬 분해\n",
    "\n",
    "$$A = X{Y}^{T}$$\n",
    "\n",
    "- 큰 행렬 A를 더 작은 행렬 X와 Y의 행렬곱으로 분해\n",
    "- X와 Y는 매우 많은 행을 가지게 되지만 열은 몇 개(k) 되지 않음.\n",
    "- k개의 열은 상호작용하는 데이터를 설명하는데 사용하는 잠재 요인에 해당됨.\n",
    "\n",
    "** [ ex ] **\n",
    "<img src = http://data-artisans.com/img/blog/factorization.svg>\n",
    "\n",
    "\n",
    "### 장점\n",
    "- A는 매우 희소(sparse)한데 비해 **XY^T**는 밀도가 매우 높아서 행렬 A에서 결측된 많은 원소에 대한 값조차 생성해줌.\n",
    "\n",
    "### 단점\n",
    "- A를 완벽하게 표현하기에는 X와 Y가 충분히 크지 않아 **A = XY^T**의 일반적인 해를 구할 수 없다."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## (2) 교차 최소 제곱 (ALS, Alternating Least Squares)\n",
    "- XY^T는 A에 가능한 한 가까워야함.\n",
    "- Y값은 모르지만, 무작위로 값이 선택된 행 벡터로 초기화 가능.\n",
    "- 그런 다음 A와 Y에 대한 최적 X를 구할 수 있음.\n",
    "- X의 각 행 i는 Y와 A의 한 행의 함수로 독립적으로 쉽게 계산할 수 있음. 독립적으로 수행할 수 있으니 병렬처리가 가능하며, 대규모로 계산할 때 매우 훌륭한 장점으로 작용.\n",
    "\n",
    "(1) $${A}_{i}Y{({Y}^{T}Y)}^{-1} = {X}_{i}$$\n",
    "\n",
    "(2) $$|{A}_{i}Y{({Y}^{T}Y)}^{-1} - {X}_{i}|$$\n",
    "\n",
    "- 하지만, (1)의 양 변을 똑같이 만들기란 불가능 하므로 (2)를 최소화 시켜야함.\n",
    "-> 여기서 Least Squares란 이름이 유래됨\n",
    "\n",
    "### 장점\n",
    "- 입력 데이터 희소성의 장점을 잘 살릴 수 있고, 간단하고 최적화된 선형대수 기법과 이 기법이 데이터 중심의 병렬 처리에 의존한다는 점 덕분에 큰 규모에서도 매우 빠르게 데이터를 처리할 수 있음."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. 데이터 준비하기\n",
    "- http://spark.apache.org/docs/latest/ml-guide.html\n",
    "- Spark 2.x 부터는 spark.ml의 DataFrame-based API를 주요 machine learning API로 사용함"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "base = \"profiledata_06-May-2005/\"\n",
    "rawUserArtistData = sc.textFile(base + \"user_artist_data.txt\")\n",
    "rawArtistData = sc.textFile(base + \"artist_data.txt\")\n",
    "rawArtistAlias = sc.textFile(base + \"artist_alias.txt\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## (1) rawUserArtistData ( user_artist_data.txt )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1000002 1 55', '1000002 1000006 33', '1000002 1000007 8']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 사용자 ID, 아티스트ID, 재생 횟수\n",
    "rawUserArtistData.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "userArtistDf = rawUserArtistData.map(lambda x: [int(a) for a in x.split(\" \")])\\\n",
    ".toDF([\"user\", \"artist\", \"playtime\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+--------+\n",
      "|   user| artist|playtime|\n",
      "+-------+-------+--------+\n",
      "|1000002|      1|      55|\n",
      "|1000002|1000006|      33|\n",
      "|1000002|1000007|       8|\n",
      "|1000002|1000009|     144|\n",
      "|1000002|1000010|     314|\n",
      "|1000002|1000013|       8|\n",
      "|1000002|1000014|      42|\n",
      "|1000002|1000017|      69|\n",
      "|1000002|1000024|     329|\n",
      "|1000002|1000025|       1|\n",
      "|1000002|1000028|      17|\n",
      "|1000002|1000031|      47|\n",
      "|1000002|1000033|      15|\n",
      "|1000002|1000042|       1|\n",
      "|1000002|1000045|       1|\n",
      "|1000002|1000054|       2|\n",
      "|1000002|1000055|      25|\n",
      "|1000002|1000056|       4|\n",
      "|1000002|1000059|       2|\n",
      "|1000002|1000062|      71|\n",
      "+-------+-------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "userArtistDf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+------------------+------------------+\n",
      "|summary|              user|            artist|          playtime|\n",
      "+-------+------------------+------------------+------------------+\n",
      "|  count|          24296858|          24296858|          24296858|\n",
      "|   mean|1947573.2653533637|1718704.0937568964| 15.29576248089362|\n",
      "| stddev| 496000.5551818977|2539389.0924283406|153.91532446980173|\n",
      "|    min|                90|                 1|                 1|\n",
      "|    max|           2443548|          10794401|            439771|\n",
      "+-------+------------------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 요약 통계\n",
    "userArtistDf.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('user', 'bigint'), ('artist', 'bigint'), ('playtime', 'bigint')]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "userArtistDf.dtypes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## (2) rawArtistData ( artist_data.txt )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1134999\\t06Crazy Life',\n",
       " '6821360\\tPang Nakarin',\n",
       " '10113088\\tTerfel, Bartoli- Mozart: Don',\n",
       " '10151459\\tThe Flaming Sidebur',\n",
       " '6826647\\tBodenstandig 3000',\n",
       " '10186265\\tJota Quest e Ivete Sangalo',\n",
       " '6828986\\tToto_XX (1977',\n",
       " '10236364\\tU.S Bombs -',\n",
       " '1135000\\tartist formaly know as Mat',\n",
       " '10299728\\tKassierer - Musik für beide Ohren']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 사용자 ID, 아티스트 이름\n",
    "rawArtistData.take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['1134999', '06Crazy Life'],\n",
       " ['6821360', 'Pang Nakarin'],\n",
       " ['10113088', 'Terfel, Bartoli- Mozart: Don'],\n",
       " ['10151459', 'The Flaming Sidebur'],\n",
       " ['6826647', 'Bodenstandig 3000'],\n",
       " ['10186265', 'Jota Quest e Ivete Sangalo'],\n",
       " ['6828986', 'Toto_XX (1977'],\n",
       " ['10236364', 'U.S Bombs -'],\n",
       " ['1135000', 'artist formaly know as Mat'],\n",
       " ['10299728', 'Kassierer - Musik für beide Ohren']]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rawArtistData.map(lambda line: line.split(\"\\t\")).take(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 탭을 포함하지 않거나 의도치 않은 개행 문자가 포함될 경우 파싱이 제대로 되지 않을 수있음."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[14] at RDD at PythonRDD.scala:48"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rawArtistData.flatMap(lambda line: line.split(\"\\t\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def buildArtistByID(pair):\n",
    "    splitPair = pair.split('\\t')\n",
    "    if len(splitPair) != 2:\n",
    "        return []\n",
    "    else:\n",
    "        try:\n",
    "            return [(int(splitPair[0]), splitPair[1].strip())]\n",
    "        except:\n",
    "            return []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1134999, '06Crazy Life'),\n",
       " (6821360, 'Pang Nakarin'),\n",
       " (10113088, 'Terfel, Bartoli- Mozart: Don'),\n",
       " (10151459, 'The Flaming Sidebur'),\n",
       " (6826647, 'Bodenstandig 3000'),\n",
       " (10186265, 'Jota Quest e Ivete Sangalo'),\n",
       " (6828986, 'Toto_XX (1977'),\n",
       " (10236364, 'U.S Bombs -'),\n",
       " (1135000, 'artist formaly know as Mat'),\n",
       " (10299728, 'Kassierer - Musik für beide Ohren')]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rawArtistData.flatMap(lambda line: buildArtistByID(line)).take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "artistByID_dict = dict(rawArtistData.flatMap(lambda line: buildArtistByID(line)).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "artistByID = rawArtistData.flatMap(lambda line: buildArtistByID(line)).toDF([\"id\", \"name\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------------+\n",
      "|      id|                name|\n",
      "+--------+--------------------+\n",
      "| 1134999|        06Crazy Life|\n",
      "| 6821360|        Pang Nakarin|\n",
      "|10113088|Terfel, Bartoli- ...|\n",
      "|10151459| The Flaming Sidebur|\n",
      "| 6826647|   Bodenstandig 3000|\n",
      "|10186265|Jota Quest e Ivet...|\n",
      "| 6828986|       Toto_XX (1977|\n",
      "|10236364|         U.S Bombs -|\n",
      "| 1135000|artist formaly kn...|\n",
      "|10299728|Kassierer - Musik...|\n",
      "+--------+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "artistByID.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## (3) rawArtistAlias ( artist_alias.txt )\n",
    "- 아티스트id를 공식 아티스트id로 매핑"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1092764\\t1000311', '1095122\\t1000557', '6708070\\t1007267']"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rawArtistAlias.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def buildArtistAlias(alias):\n",
    "    splitPair = alias.split('\\t')\n",
    "    if len(splitPair) != 2:\n",
    "        return []\n",
    "    else:\n",
    "        try:\n",
    "            return [(int(splitPair[0]), int(splitPair[1]))]\n",
    "        except:\n",
    "            return []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "artistAlias = dict(rawArtistAlias.flatMap(lambda line: buildArtistAlias(line)).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# ArtistAlias = rawArtistAlias.flatMap(lambda line: buildArtistAlias(line).collect().toMap)\n",
    "\n",
    "# ArtistAlias = rawArtistAlias.flatMap(lambda line: buildArtistAlias(line).collectAsMap())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4. 모델 만들기"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.recommendation import ALS, ALSModel"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***class pyspark.ml.recommendation.ALS(self, rank=10, maxIter=10, regParam=0.1, numUserBlocks=10, numItemBlocks=10, implicitPrefs=false, alpha=1.0, userCol=\"user\", itemCol=\"item\", seed=None, ratingCol=\"rating\", nonnegative=false, checkpointInterval=10, intermediateStorageLevel=\"MEMORY_AND_DISK\", finalStorageLevel=\"MEMORY_AND_DISK\")[source]***\n",
    "\n",
    "Alternating Least Squares (ALS) matrix factorization.\n",
    "\n",
    "ALS attempts to estimate the ratings matrix R as the product of two lower-rank matrices, X and Y, i.e. X * Yt = R. Typically these approximations are called ‘factor’ matrices. The general approach is iterative. During each iteration, one of the factor matrices is held constant, while the other is solved for using least squares. The newly-solved factor matrix is then held constant while solving for the other factor matrix.\n",
    "\n",
    "This is a blocked implementation of the ALS factorization algorithm that groups the two sets of factors (referred to as “users” and “products”) into blocks and reduces communication by only sending one copy of each user vector to each product block on each iteration, and only for the product blocks that need that user’s feature vector. This is achieved by pre-computing some information about the ratings matrix to determine the “out-links” of each user (which blocks of products it will contribute to) and “in-link” information for each product (which of the feature vectors it receives from each user block it will depend on). This allows us to send only an array of feature vectors between each user block and product block, and have the product block find the users’ ratings and update the products based on these messages.\n",
    "\n",
    "For implicit preference data, the algorithm used is based on “Collaborative Filtering for Implicit Feedback Datasets”,, adapted for the blocked approach used here.\n",
    "\n",
    "Essentially instead of finding the low-rank approximations to the rating matrix R, this finds the approximations for a preference matrix P where the elements of P are 1 if r > 0 and 0 if r <= 0. The ratings then act as ‘confidence’ values related to strength of indicated user preferences rather than explicit ratings given to items."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## (1) 데이터 준비\n",
    "- 비공식 아티스트 ID를 공식 명칭의 아티스트ID로 바꿔줌.\n",
    "- 데이터를 Rating 객체로 바꾸어야됨. (사용자 - 제품 - 가치 데이터를 추상화한 것)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (1)-1 브로드캐스트"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 스파크가 어떤 단계를 실행하면 그단계에서 태스크를 실행하는데 필요한 모든 정보를 바이너리 형태로 만든다. 이 바이너리를 실행할 함수의 클로저(closure)라고 한다. 클로저는 함수가 참조하는 구동자 상의 모든 자료구조를 포함하며, 스파크는 클로저를 클러스터 상의 모든 실행자에 배포한다. 다수의 태스크가 데이터를 변경할 수 없는 (immutable) 동일한 자료구조에 접근해야 할 때는 브로드캐스트 변수를 쓰는 것이 좋다.\n",
    "\n",
    "\n",
    "- 브로드캐스트 변수의 태스크 클로저 처리 방법\n",
    "    - 각 실행자에서 데이터를 원래 자바 객체 형태로 캐시한다. 따라서 태스크마다 직렬화를 해제할 필요가 없어짐.\n",
    "    - 여러 작업과 여러 단계에 걸친 데이터를 캐시함."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "bArtistAlias = sc.broadcast(artistAlias)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- ArtistAlias ( 공식 아티스트id와 매핑 )에 연산할 때마다 접근하는 것은 비효율 적임.\n",
    "- 클러스터에 있는 excutor 하나당 딱 하나의 복사본만을 보내고 메모리에 유지하도록 할 수 있음.\n",
    "- 태스크가 수천 개쯤 되고 클러스터 상의 여러 실행자에서 병렬로 수행될 때는 이와 같은 방법으로  \n",
    "네트워크 트래픽과 메모리 사용량을 크게 줄일 수 있음."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# print \"Sampling the dataset to fit into laptop memory...\"\n",
    "# weights = [.1, .9]\n",
    "# seed = 42\n",
    "# sample, someOtherJunk = rawDataRDD.randomSplit(weights, seed)\n",
    "# sample.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def buildCounts(x):\n",
    "    line = x.split()\n",
    "#     userID, artistID, count = line.map(lambda item: int(item))\n",
    "    userID, artistID, count = [int(i) for i in line]\n",
    "    # 공식 artist id로 변환\n",
    "    finalArtistID = bArtistAlias.value.get(artistID)\n",
    "    if finalArtistID is None:\n",
    "        finalArtistID = artistID\n",
    "    Rating = (userID, finalArtistID, count)\n",
    "    return Rating"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 150,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1000002 1 55',\n",
       " '1000002 1000006 33',\n",
       " '1000002 1000007 8',\n",
       " '1000002 1000009 144']"
      ]
     },
     "execution_count": 150,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rawUserArtistData.take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[user: bigint, artist: bigint, count: bigint]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "trainData = rawUserArtistData.map(lambda x: buildCounts(x)).toDF([\"user\", \"artist\", \"count\"])\n",
    "trainData.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-----+\n",
      "|   user| artist|count|\n",
      "+-------+-------+-----+\n",
      "|1000002|      1|   55|\n",
      "|1000002|1000006|   33|\n",
      "|1000002|1000007|    8|\n",
      "|1000002|1000009|  144|\n",
      "|1000002|1000010|  314|\n",
      "|1000002|1000013|    8|\n",
      "|1000002|1000014|   42|\n",
      "|1000002|1000017|   69|\n",
      "|1000002|1000024|  329|\n",
      "|1000002|1000025|    1|\n",
      "|1000002|1000028|   17|\n",
      "|1000002|1000031|   47|\n",
      "|1000002|1000033|   15|\n",
      "|1000002|1000042|    1|\n",
      "|1000002|1000045|    1|\n",
      "|1000002|1000054|    2|\n",
      "|1000002|1000055|   25|\n",
      "|1000002|1000056|    4|\n",
      "|1000002|1000059|    2|\n",
      "|1000002|1000062|   71|\n",
      "+-------+-------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "trainData.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- ALS 알고리즘은 ?? 데이터를 읽어야 하며 보통 10회 이상 반복 수행하기 때문에 캐시하는 쪽이 바람직함\n",
    "- 캐시하지 않더라도 데이터가 필요할 떄마다 원래 데이터로부터 다시 계산하여 만드는 일을 반복할 수 있음"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "model = ALS(rank = 10, regParam=0.1, maxIter=5, alpha=10, implicitPrefs=True, userCol='user'\n",
    "    , itemCol='artist', ratingCol='count').fit(trainData)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- ALS 모델은 사용자와 상품 각각에 대해서 10개의 값으로 이루어진 특징 벡터 (feature vector)를 가지며, 여기에서는 이런 벡터가 170만 개 이상 만들어진다."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+\n",
      "| id|            features|\n",
      "+---+--------------------+\n",
      "| 30|[0.054098744, -0....|\n",
      "| 40|[0.009872092, 0.0...|\n",
      "| 50|[0.016281795, 0.0...|\n",
      "| 70|[0.031204568, 0.0...|\n",
      "| 90|[-0.2005218, 0.08...|\n",
      "|100|[0.03972331, 0.04...|\n",
      "|120|[0.06705716, 0.04...|\n",
      "|140|[0.0057025356, -2...|\n",
      "|150|[0.066728376, -0....|\n",
      "|170|[3.297552E-4, -0....|\n",
      "|180|[0.014222841, -0....|\n",
      "|190|[-1.740271E-4, -3...|\n",
      "|200|[0.015938388, 0.1...|\n",
      "|210|[0.0039446596, 0....|\n",
      "|230|[-0.0050065895, 0...|\n",
      "|240|[0.033835486, -0....|\n",
      "|250|[0.0780018, -0.00...|\n",
      "|280|[0.0077589205, -0...|\n",
      "|290|[-0.038641848, -0...|\n",
      "|300|[0.06371394, -0.0...|\n",
      "+---+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# item factor\n",
    "model.itemFactors.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------------+\n",
      "|  id|            features|\n",
      "+----+--------------------+\n",
      "|  90|[0.24294847, 0.36...|\n",
      "| 120|[0.15089737, 0.04...|\n",
      "| 340|[-0.0016096301, 0...|\n",
      "| 350|[0.38526383, -0.1...|\n",
      "| 770|[-0.14820945, 0.0...|\n",
      "|3290|[1.7071291, -0.28...|\n",
      "|4370|[0.3261275, -0.08...|\n",
      "|4620|[0.08121199, 0.00...|\n",
      "|6060|[0.44454375, 0.38...|\n",
      "|6390|[-0.17291078, -0....|\n",
      "|6760|[0.4391811, 0.207...|\n",
      "|6850|[0.5124266, 0.066...|\n",
      "|7010|[0.51738626, 0.43...|\n",
      "|7130|[0.81357825, 0.67...|\n",
      "|7290|[0.42321518, 0.07...|\n",
      "|7340|[0.7557541, 0.228...|\n",
      "|7400|[0.9815967, -0.58...|\n",
      "|7510|[0.27876863, -0.0...|\n",
      "|8500|[0.014992191, 0.0...|\n",
      "|9660|[0.47934183, 0.43...|\n",
      "+----+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# user factor\n",
    "model.userFactors.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 5. 추천 결과 확인"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "- testId = 2093760\n",
    "- 사용자가 청취한 아티스트들의 ID를 뽑아 출력\n",
    "- 이 사용자가 들은 아티스트들의 ID를 찾고, 이 ID들을 사용해 아티스트 집합을 필터링하고,  \n",
    "그 결과를 모아 아티스트 이름을 순서대로 출력"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+--------+\n",
      "|   user| artist|playtime|\n",
      "+-------+-------+--------+\n",
      "|2093760|   1180|       1|\n",
      "|2093760|1255340|       3|\n",
      "|2093760|    378|       1|\n",
      "|2093760|    813|       2|\n",
      "|2093760|    942|       7|\n",
      "+-------+-------+--------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "DataFrame[user: bigint, artist: bigint, playtime: bigint]"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "userArtistDf.filter(userArtistDf.user == 2093760).show()\n",
    "userArtistDf.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+\n",
      "|           name|\n",
      "+---------------+\n",
      "|  Blackalicious|\n",
      "|     Jurassic 5|\n",
      "|The Saw Doctors|\n",
      "|         Xzibit|\n",
      "|     David Gray|\n",
      "+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 2093760 유저가 청취했던 아티스트\n",
    "\n",
    "artistByID.join(userArtistDf.filter(userArtistDf.user == 2093760).select('artist').withColumnRenamed('artist', 'id'), \n",
    "                on = 'id').select('name').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 힙합 & 팝 가수"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def makeRecommendation(model, userId, howMany):\n",
    "    from pyspark.sql.functions import lit\n",
    "    # The passed in object is returned directly if it is already a Column. If the object is a Scala Symbol, it is converted into a Column also. Otherwise, a new Column is created to represent the literal value.\n",
    "    # withcolumn 은 column 형태만 가능함\n",
    "    toRecommend = model.itemFactors.withColumnRenamed('id',\"artist\").select('artist').withColumn(\"user\", lit(userId))\n",
    "    result = model.transform(toRecommend).select('artist', 'prediction').orderBy('prediction', ascending = False).limit(howMany)\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 추천 결과\n",
    "def recommenction(model, userId, howMany):\n",
    "    topRecommecdation = makeRecommendation(model, userId, howMany)\n",
    "    recommendationArtistID = topRecommecdation.select(\"artist\").collect()\n",
    "    return artistByID.join(spark.createDataFrame(recommendationArtistID).toDF('id'), on = 'id').select('name').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+\n",
      "|        name|\n",
      "+------------+\n",
      "|Gwen Stefani|\n",
      "|   blink-182|\n",
      "| The Killers|\n",
      "|      Eminem|\n",
      "|     50 Cent|\n",
      "+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 추천 결과\n",
    "recommenction(model, 2093760, 5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 6. 추천 품질 평가하기"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## (1) 평가 스코어\n",
    "\n",
    "- **FDS(Fraud Detection System)의 예**\n",
    "    - FDS(Fraud Detection System)는 금융 거래, 회계 장부 등에서 잘못된 거래, 사기 거래를 찾아내는 시스템을 말한다. FDS의 예측 결과가 Positive 이면 사기 거래라고 예측한 것이고 Negative 이면 정상 거래라고 예측한 것이다. 이 결과가 사실과 일치하는지 틀리는지에 따라 다음과 같이 말한다.\n",
    "\n",
    "        - **True Positive**: 사기를 사기라고 정확하게 예측\n",
    "        - **True Negative**: 정상을 정상이라고 정확하게 예측\n",
    "        - **False Positive**: 정상을 사기라고 잘못 예측\n",
    "        - **False Negative**: 사기를 정상이라고 잘못 예측\n",
    "\n",
    "\n",
    "$$ \\begin{vmatrix}\n",
    "\\cdots & 사기 거래라고 예측 & 정상 거래라고 예측\\\\\n",
    "실제로 사기 거래 & True Positive & False Negative\\\\\n",
    "실제로 정상 거래 & False Positive & True Negative \n",
    "\\end{vmatrix} $$\n",
    "\n",
    "\n",
    "### Accuracy 정확도\n",
    "- 전체 샘플 중 맞게 출력한 샘플 수의 비율\n",
    "\n",
    "$$ accuracy = \\frac{TP + TN}{TP + TN + FP + FN} $$\n",
    "\n",
    "### Precision 정밀도\n",
    "- 클래스에 속한다고 출력한 샘플 중 실제로 클래스에 속하는 샘플 수의 비율\n",
    "- FDS의 경우, 사기 거래라고 판단한 거래 중 실제 사기 거래의 시율. 유죄율\n",
    "\n",
    "$$ precision = \\frac{TP}{TP + FP} $$\n",
    "\n",
    "\n",
    "### Recall 재현율\n",
    "- TPR : true positive rate\n",
    "- 실제 클래스에 속한 샘플 중에 클래스에 속한다고 출력한 샘플의 수\n",
    "- FDS의 경우, 실제 사기 거래 중에서 실제 사기 거래라고 예측한 거래의 비율. 검거율\n",
    "- sensitivity (민감도)\n",
    "\n",
    "$$ recall = \\frac{TP}{TP + FN} $$\n",
    "\n",
    "\n",
    "### Fall - Out\n",
    "- FPR : false positive rate\n",
    "- 실제 클래스에 속하지 않는 샘플 중에 클래스에 속한다고 출력한 샘플의 수\n",
    "- FDS의 경우, 실제 정상 거래 중에서 FDS가 사기 거래라고 예측한 거래의 비율\n",
    "\n",
    "$$ fallout = \\frac{FP}{FP + TN} $$\n",
    "\n",
    "\n",
    "### F (beta) score\n",
    "- 정밀도 (Precision)과 재현율 (Recall)의 가중 조화 평균\n",
    "\n",
    "$$ F_{\\beta} = \\frac{(1 + \\beta^2)(precision * recall)}{\\beta^2 precision + recall} $$\n",
    "\n",
    "\n",
    "- F1 score\n",
    "    - beta = 1\n",
    "        $$ F_{1} = \\frac{2*precision * recall}{precision + recall} $$\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## (2) ROC (Receiver Operating Characteristic)\n",
    "\n",
    "- **ROC(Receiver Operator Characteristic) 커브**는 클래스 판별 기준값의 변화에 따른 Fall-out과 Recall의 변화를 시각화한 것이다.  \n",
    "모든 이진 분류 모형은 판별 평면으로부터의 거리에 해당하는 판별 함수(discriminant function)를 가지며 판별 함수 값이 음수이면 0인 클래스, 양수이면 1인 클래스에 해당한다고 판별한다. 즉 0 이 클래스 판별 기준값이 된다. ROC 커브는 이 클래스 판별 기준값이 달라진다면 판별 결과가 어떻게 달라지는지는 표현한 것이다.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** [ ex ] **\n",
    "<img src = http://scikit-learn.org/stable/_images/sphx_glr_plot_roc_001.png>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## (3) AUC (Area Under the Curve)\n",
    "- AUC는 ROC curve의 면접을 뜻한다. Fall-Out 대비 Recall 값이 클 수록 AUC가 1에 가까운 값이며 우수한 모형이다."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# 모델 평가 부분은 추후 업데이트 하겠습니다."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# 7. 하이퍼파라미터 선택하기"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### (1) rank = 10\n",
    "    - 모델의 잠재 요긴 개수. 즉, 사용자 - 특징 행렬과 제품 - 특징 행렬에서 열의 개수 k. 자명하지 않은 사용자 - 특징 또는 제품 - 특징 행렬을 가지는 경우, 이 값은 각 행렬의 계수(rank)이기도 함.\n",
    "    \n",
    "### (2) maxIter = 5\n",
    "    - 행렬 분해를 반복하는 횟수. 반복 횟수가 많을수록 더 긴 시간이 필요하지만, 행렬 분해 결과는 더 좋아질 것이다.\n",
    "\n",
    "### (3) regParam = 0.01\n",
    "    - overfitting 파라미터. 높은 값일 수록 overfitting을 피하기 쉽지만, 너무 큰 값을 부여하면 행렬 분해의 정확도가 떨어질 수 있다.\n",
    "\n",
    "### (4) alpha = 10\n",
    "    - 행렬 분해 과정에서 사용자 - 제품 상호작용 중 관측치와 비관측치의 상대적인 가중치를 조절한다."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [conda root]",
   "language": "python",
   "name": "conda-root-py"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}