Project1_Sneha Ganganna&Aishwarya Dinni.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "b984f2be-433a-416b-a6bf-512b8649f076",
   "metadata": {},
   "source": [
    "# Advanced Database and Information Systems\n",
    "## Project 1\n",
    "### Sneha Ganganna (5579362), Aishwarya Dinni (5653414)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "d96c8db7-6b18-4bb4-9ffc-dfcd2d116192",
   "metadata": {},
   "outputs": [],
   "source": [
    "#To check if the spark has been installed \n",
    "#!pip install pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "c2bd3028-a5ec-4cb8-a30d-5ffb7346828e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import csv\n",
    "import time\n",
    "import pyspark\n",
    "from pyspark import SparkConf\n",
    "from pyspark.sql.types import *\n",
    "from collections import Counter\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.functions import row_number\n",
    "from pyspark.sql.functions import countDistinct"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "424c909e-406c-4c18-915d-2be07af2546d",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# initialise Spark Session\n",
    "sparkSession = SparkSession.builder.appName(\"exe1\").getOrCreate()\n",
    "sc = sparkSession.sparkContext\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "251fe22b-ab09-4a41-b10a-70e9b8cf91aa",
   "metadata": {},
   "source": [
    "# Exercise 1. 2 (Loading the dataset into an RDD)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "80f50c31-e92a-495f-9c28-e354956c6b46",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "# Read users_libraries file into RDD\n",
    "usersRatingsRDD = sc.textFile(\"users_libraries\")\n",
    "# Create Pair RDD for users_libraries Text File\n",
    "usersRatingsRDD = sc.textFile(\"users_libraries.txt\").map(lambda line: line.split(\";\")).map(lambda line: (line[0],list(map(int,line[1].split(\",\")))))\n",
    "\n",
    "\n",
    "#Stopword Broadcast\n",
    "stopWords = sc.textFile(\"stopwords_en.txt\")\n",
    "stopWordsBroadcast = sc.broadcast(stopWords.collect())\n",
    "\n",
    "\n",
    "#PaperCsv RDD\n",
    "def processingCsvFile(line):\n",
    "    papers = csv.reader([line.replace(\"\\0\", \"\")], delimiter=',', quoting=csv.QUOTE_MINIMAL)\n",
    "    paperList = next(papers)\n",
    "    return paperList[0], paperList[14]\n",
    "\n",
    "papersTermsRDD = sc.textFile(\"papers.csv\")\n",
    "papersTermsRDD = papersTermsRDD.map(processingCsvFile).filter(lambda x: (x[1] != \"\" and x[1] != \" \")).map(lambda x: (int(x[0]),x[1].split(\" \")))\n",
    "\n",
    "# Join the two RDDs on the common key i.e. paper_id\n",
    "papersTerms_userRatings = usersRatingsRDD.flatMapValues(lambda x: x).map(lambda x: (x[1],x[0])).join(papersTermsRDD)\n",
    "\n",
    "# Print the result\n",
    "papersTerms_userRatings.foreach(print)\n",
    "papersTerms_userRatings.foreach(lambda x: print(x))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dadcc09e-48ca-42c4-ac8f-7c4fba573506",
   "metadata": {},
   "source": [
    "# Exercise 1. 3 (Joining collections)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "8a3a990d-9ec8-40ad-8186-004a5498fcac",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Execution Time taken to Join collections: 0.03113842010498047 in seconds\n"
     ]
    }
   ],
   "source": [
    "start_time = time.time()\n",
    "\n",
    "# Finding top 10 most frequent words\n",
    "def FrequentWords(x):\n",
    "    rList = Counter(x)\n",
    "    FrequentWordsCount = CounterList.most_common(10)\n",
    "    FrequentWords = [word for word, word_count in FrequentWordsCount]\n",
    "    return FrequentWords\n",
    "\n",
    "#to Remove the stopWords\n",
    "def removeStopWords(List):\n",
    "    requiredList = List.copy()\n",
    "    for x in List:\n",
    "        if ((x in stopWordsBroadcast.value) or x == \"\" or x == \" \"):\n",
    "            requiredList.remove(x)\n",
    "    return requiredList\n",
    "\n",
    "\n",
    "# join the RDDs userLibraries and PapersRdd\n",
    "usersLibraries_PapersRdd = usersRatingsRDD.flatMapValues(lambda x: x).map(lambda x: (x[1],x[0])).join(papersTermsRDD)\n",
    "#restructure the RDD\n",
    "usersLibraries_PapersRdd = usersLibraries_PapersRdd.map(lambda x: (x[1][0],x[1][1]))\n",
    "#transformation of RDD to perform computations \n",
    "usersLibraries_PapersRdd   = usersLibraries_PapersRdd.flatMapValues(lambda x:x).groupByKey().mapValues(list)\n",
    "#to remove the stop words from the RDD\n",
    "#let Rdd be usersLibraries_PapersRdd\n",
    "rddWithoutStopWords = usersLibraries_PapersRdd.mapValues(removeStopWords)\n",
    "\n",
    "# now run the def frequentWords after the stop words are removed\n",
    "frequentWordList = rddWithoutStopWords.mapValues(FrequentWords)\n",
    "\n",
    "#wrtiting the above data into file\n",
    "def CreateCsv(data):\n",
    "    Data = data[0] + \",\" + (','.join(str(d) for d in data[1]))\n",
    "    return Data\n",
    "\n",
    "frequentWordListFile = frequentWordList.map(CreateCsv)\n",
    "#frequentWordListFile.saveAsTextFile(\"result/Results\")\n",
    "\n",
    "end_time = time.time()\n",
    "\n",
    "print(\"Execution Time taken to Join collections:\", end_time - start_time, \"in seconds\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f9a1ebc6-6d23-47ad-8d7c-ed5c4a41357b",
   "metadata": {},
   "source": [
    "# Exercise 1. 4 (Basic Analysis for Recommender Systems)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "97614e8a-34ae-46ce-9bfb-48266ecf08cc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of distinct users:  28416\n",
      "Number of distinct items:  172079\n",
      "Number of ratings:  828481\n",
      "Minimum number of ratings a user has given:  1\n",
      "Maximum number of ratings a user has given:  1922\n",
      "Average number of ratings a user has given:  29.155440596846848\n",
      "Standard deviation of ratings for users 81.1751761366871\n",
      "Minimum number of ratings an item has received:  3\n",
      "Maximum number of ratings an item has received:  924\n",
      "Average number of ratings per item:  4.81453867119172\n",
      "Standard deviation of ratings per item:  5.477802292314525\n",
      "The time taken for the Basic Analysis of Recommender Systems: 6.037190675735474 seconds\n"
     ]
    }
   ],
   "source": [
    "#to record execution time\n",
    "start_time = time.time()\n",
    "\n",
    "#a) Number of (distinct) user, number of (distinct) items, and number of ratings\n",
    "\n",
    "#first apply flatMapVales() function for each kay-pair value in userRatingsRDD\n",
    "usersRatingsRDDFMV =usersRatingsRDD.flatMapValues(lambda x:x)\n",
    "\n",
    "numOfUsers = usersRatingsRDDFMV.keys().distinct().count()\n",
    "print(\"Number of distinct users: \", numOfUsers)\n",
    "\n",
    "numOfItems = usersRatingsRDDFMV.values().distinct().count()\n",
    "print(\"Number of distinct items: \", numOfItems)\n",
    "\n",
    "numOfRatings = usersRatingsRDDFMV.count()\n",
    "print(\"Number of ratings: \", numOfRatings)\n",
    "\n",
    "#b) Min number of ratings a user has given\n",
    "minNumberOfRatings = usersRatingsRDD.map(lambda x: (x[0],len(x[1]))).map(lambda x: x[1]).min()\n",
    "print(\"Minimum number of ratings a user has given: \", minNumberOfRatings)\n",
    "\n",
    "#c)  Max number of ratings a user has given\n",
    "maxNumberOfRatings = usersRatingsRDD.map(lambda x: (x[0],len(x[1]))).map(lambda x: x[1]).max()\n",
    "print(\"Maximum number of ratings a user has given: \", maxNumberOfRatings)\n",
    "\n",
    "#d)  Average number of ratings of users\n",
    "avgNumberOfRatings = numOfRatings/numOfUsers\n",
    "print(\"Average number of ratings a user has given: \", avgNumberOfRatings)\n",
    "\n",
    "#e) Standard deviation for ratings of users\n",
    "stdevRatingsOfUsers = (usersRatingsRDD.map(lambda x: (x[0],len(x[1]))).map(lambda x: x[1])).stdev()\n",
    "print(\"Standard deviation of ratings for users\",stdevRatingsOfUsers)\n",
    " \n",
    "#f)Min number of ratings an item has received\n",
    "minRatingsOfitem = usersRatingsRDDFMV.map(lambda x: (x[1],1)).reduceByKey(lambda x,y: x+y).map(lambda x: x[1]).min()\n",
    "print(\"Minimum number of ratings an item has received: \", minRatingsOfitem)\n",
    "\n",
    "#g) Max number of ratings an item has received\n",
    "maxRatingsOfitem = usersRatingsRDDFMV.map(lambda x: (x[1],1)).reduceByKey(lambda x,y: x+y).map(lambda x: x[1]).max()\n",
    "print(\"Maximum number of ratings an item has received: \", maxRatingsOfitem)\n",
    "\n",
    "#h) Average number of ratings of items\n",
    "avgRatingsOfitem= numOfRatings/numOfItems\n",
    "print(\"Average number of ratings per item: \", avgRatingsOfitem)\n",
    "\n",
    "#i) Standard deviation for ratings of items\n",
    "stdevRatingsOfItems = usersRatingsRDDFMV.map(lambda x: (x[1],1)).reduceByKey(lambda x,y: x+y).map(lambda x: x[1]).stdev()\n",
    "print(\"Standard deviation of ratings per item: \", stdevRatingsOfItems)\n",
    "\n",
    "end_time = time.time()\n",
    "print(\"The time taken for the Basic Analysis of Recommender Systems:\", end_time-start_time,\"seconds\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "54592776-5de0-4dd5-a855-311f305b30c0",
   "metadata": {},
   "source": [
    "# Exercise 1. 5 (Loading the dataset into Data Frames)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "b200ff62-ed91-4a25-bd1e-c6394c668229",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(paper_id=5842862, type='article', journal='molecular cell', bookـtitle=None, series=None, publisher='elsevier', pages=2, volume=35, number=6, year=2009, month='sep', postedat=None, address=None, title='how to choose a good scientific problem', abstract='choosing good problems is essential for being a good scientist. but what is a good problem, and how do you choose one? the subject is not usually discussed explicitly within our profession. scientists are expected to be smart enough to figure it out on their own and through the observation of their teachers. this lack of explicit discussion leaves a vacuum that can lead to approaches such as choosing problems that can give results that merit publication in valued journals, resulting in a job and tenure.')"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# To define the schema of the CSV data\n",
    "papersSchema = StructType([\n",
    "    StructField(\"paper_id\", IntegerType(), True),\n",
    "    StructField(\"type\", StringType(), True),\n",
    "    StructField(\"journal\", StringType(), True),\n",
    "    StructField(\"bookـtitle\", StringType(), True),\n",
    "    StructField(\"series\", IntegerType(), True),\n",
    "    StructField(\"publisher\", StringType(), True),\n",
    "    StructField(\"pages\", IntegerType(), True),\n",
    "    StructField(\"volume\", IntegerType(), True),\n",
    "    StructField(\"number\", IntegerType(), True),\n",
    "    StructField(\"year\", IntegerType(), True),\n",
    "    StructField(\"month\", StringType(), True),\n",
    "    StructField(\"postedat\", FloatType(), True),\n",
    "    StructField(\"address\", StringType(), True),\n",
    "    StructField(\"title\", StringType(), True),\n",
    "    StructField(\"abstract\", StringType(), True)\n",
    "    ])\n",
    "papersCsvDf = sparkSession.read.csv(\"papers.csv\", sep = \",\", header = True, schema = papersSchema)\n",
    "papersCsvDf = papersCsvDf.na.drop(subset=[\"abstract\"]) # drops the null values\n",
    "\n",
    "\n",
    "\n",
    "# To define the schema of the text data\n",
    "userLibrariesSchema = StructType([\n",
    "    StructField(\"user_hash_id\",StringType(),True),\n",
    "    StructField(\"user_library\",StringType(),True)\n",
    "])\n",
    "usersLibrariesDf = sparkSession.read.csv(\"users_libraries.txt\", sep = \";\", header = True, schema = userLibrariesSchema)\n",
    "# Select the columns using selectExpr() function\n",
    "usersLibrariesDf = usersLibrariesDf.selectExpr(\"user_hash_id\",\"split(user_library,',') AS user_library\")\n",
    "\n",
    "usersLibrariesDf.head()\n",
    "\n",
    "papersCsvDf.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4a07d744-62b9-4873-a5db-5fd67dbf822c",
   "metadata": {},
   "source": [
    "# Exercise 1. 6 Tasks on top of DataFrames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "cdab27f4-8288-4b19-93a7-237bf238a8e9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Execution Time taken to Join collections: 0.34844231605529785 seconds\n"
     ]
    }
   ],
   "source": [
    "#to record execution time\n",
    "startTime = time.time()\n",
    "\n",
    "#Explode UserLibrary\n",
    "usersLibrariesExplode = usersLibrariesDf.select(usersLibrariesDf.user_hash_id,explode(usersLibrariesDf.user_library).alias(\"paper_id\"))\n",
    "\n",
    "#Join UserLibrary and PaperCsv\n",
    "usersLibraries_PapersCsvDf = papersCsvDf.join(usersLibrariesExplode,papersCsvDf\\\n",
    "                                              .paper_id == usersLibrariesExplode.paper_id, how=\"inner\")\\\n",
    "                                            .select(usersLibrariesExplode.user_hash_id,usersLibrariesExplode.paper_id,papersCsvDf.abstract)\n",
    "\n",
    "\n",
    "#Removing stop words\n",
    "# Split the abstract column into an array of words\n",
    "usersLibraries_PapersCsvDf = usersLibraries_PapersCsvDf.withColumn(\"abstract_words\", split(usersLibraries_PapersCsvDf.abstract, \" \"))\n",
    "\n",
    "# Explode the abstract_words array\n",
    "usersLibraries_PapersCsvDf = usersLibraries_PapersCsvDf.select(usersLibraries_PapersCsvDf.user_hash_id, explode(usersLibraries_PapersCsvDf.abstract_words).alias(\"abstract\"))\n",
    "uselessWords = ['',' ','\"']\n",
    "#just to make it easy -  df=usersLibraries_PapersCsvDf\n",
    "dfWithoutStopWords = usersLibraries_PapersCsvDf[~usersLibraries_PapersCsvDf[\"abstract\"].isin(stopWordsBroadcast.value)]\n",
    "dfWithoutStopWords = dfWithoutStopWords[~dfWithoutStopWords[\"abstract\"].isin(uselessWords)]\n",
    "\n",
    "#Finding top 10 most frequent words\n",
    "dfWithoutStopWordsCount = dfWithoutStopWords.groupBy(\"user_hash_id\",\"abstract\").count().withColumnRenamed(\"count\", \"word_count\")\n",
    "usersWords = Window.partitionBy(dfWithoutStopWordsCount.user_hash_id).orderBy(col(\"word_count\").desc())\n",
    "dfWithoutStopWordsRank = dfWithoutStopWordsCount.withColumn(\"word_rank\",rank().over(usersWords))\n",
    "topFrequentWordsPerUserDf = dfWithoutStopWordsRank.filter(dfWithoutStopWordsRank[\"word_rank\"]<11)\n",
    "groupTop10FrequentWordsPerUserDf = topFrequentWordsPerUserDf.groupBy(\"user_hash_id\").agg(collect_list(\"abstract\")).withColumnRenamed(\"collect_list(abstract)\", \"abstract_word_list\")\n",
    "\n",
    "#writing top 10 most frequent words of each user to file\n",
    "#groupTop10FrequentWordsPerUserDf.write.save(\"Top10WordsForEachUserDF\")\n",
    "\n",
    "endTime = time.time()\n",
    "\n",
    "print(\"Execution Time taken to Join collections:\", endTime-startTime, \"seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "e170dc3d-d372-4a28-a12f-c37070506691",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of (distinct) user: 28415\n",
      "Number of (distinct) items: 172079\n",
      "Number of ratings: 828461\n",
      "Minimum number of ratings a user has given:  1\n",
      "Min number of ratings a user has given: 1922\n",
      "Average number of ratings a user has given:  29.15576280133732\n",
      "Standard deviation of ratings for users 81.17801478819557\n",
      "Min number of ratings an item has received: 2\n",
      "Max number of ratings an item has received: 924\n",
      "Average number of ratings of items: 4.814422445504681\n",
      "Standard deviation for ratings of items:  5.477832307606936\n",
      "The time taken for the fundamental analysis of recommender systems: 3.0855443477630615 seconds\n"
     ]
    }
   ],
   "source": [
    "#to record execution time\n",
    "startTime = time.time()\n",
    "\n",
    "#a) Number of (distinct) user, number of (distinct) items, and number of ratings\n",
    "numOfDistinctUsersDf = str(usersLibrariesExplode.select(countDistinct(\"user_hash_id\")).collect()[0][0])\n",
    "print(\"Number of (distinct) user:\" ,numOfDistinctUsersDf)\n",
    "\n",
    "numOfDistinctItemsDf = str(usersLibrariesExplode.select(countDistinct(\"paper_id\")).collect()[0][0])\n",
    "print(\"Number of (distinct) items:\" ,numOfDistinctItemsDf)\n",
    "\n",
    "numOfRatingsDf = usersLibrariesExplode.count()\n",
    "print(\"Number of ratings:\" ,numOfRatingsDf)\n",
    "\n",
    "\n",
    "ratingsListDf = usersLibrariesExplode.groupBy(\"user_hash_id\").count().withColumnRenamed(\"count\",\"no_of_items\")\n",
    "ratingsListDf = ratingsListDf.describe(\"no_of_items\")\n",
    "\n",
    "#b) Min number of ratings a user has given\n",
    "minNumOfRatingsDf = str(ratingsListDf.filter(\"summary == 'min'\").collect()[0][1])\n",
    "print(\"Minimum number of ratings a user has given: \", minNumOfRatingsDf)\n",
    "\n",
    "#c)  Max number of ratings a user has given\n",
    "maxNumOfRatingsDf = str(ratingsListDf.filter(\"summary == 'max'\").collect()[0][1])\n",
    "print(\"Min number of ratings a user has given:\",maxNumOfRatingsDf)\n",
    "\n",
    "#d)  Average number of ratings of users\n",
    "avgNumOfRatings = int(numOfRatingsDf)/int(numOfDistinctUsersDf)\n",
    "print(\"Average number of ratings a user has given: \", avgNumOfRatings)\n",
    "\n",
    "#e) Standard deviation for ratings of users\n",
    "stdevRatingsOfUsers = str(ratingsListDf.filter(\"summary == 'stddev'\").collect()[0][1])\n",
    "print(\"Standard deviation of ratings for users\",stdevRatingsOfUsers)\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "ratingsListByPaperIdDf = usersLibrariesExplode.groupBy(\"paper_id\").count().withColumnRenamed(\"count\",\"no_of_ratings\")\n",
    "ratingsListByPaperIdDf = ratingsListByPaperIdDf.describe(\"no_of_ratings\")\n",
    "\n",
    "\n",
    "#h) Average number of ratings of items\n",
    "minNumOfRatingItemGotDf = str(ratingsListByPaperIdDf.filter(\"summary == 'min'\").collect()[0][1])\n",
    "print(\"Min number of ratings an item has received:\",minNumOfRatingItemGotDf)\n",
    "\n",
    "#g) Max number of ratings an item has received\n",
    "maxNumOfRatingItemGotDf = str(ratingsListByPaperIdDf.filter(\"summary == 'max'\").collect()[0][1])\n",
    "print(\"Max number of ratings an item has received:\",maxNumOfRatingItemGotDf)\n",
    "\n",
    "#h) Average number of ratings of items\n",
    "avgNumOfRatingOfItemsDf = int(numOfRatingsDf)/int(numOfDistinctItemsDf)\n",
    "print(\"Average number of ratings of items:\",avgNumOfRatingOfItemsDf)\n",
    "\n",
    "#i) Standard deviation for ratings of items\n",
    "stdevRatingsOfItemsDf = str(ratingsListByPaperIdDf.filter(\"summary == 'stddev'\").collect()[0][1])\n",
    "print(\"Standard deviation for ratings of items: \", stdevRatingsOfItemsDf)\n",
    "\n",
    "endTime = time.time()\n",
    "\n",
    "print(\"The time taken for the fundamental analysis of recommender systems:\", endTime-startTime,\"seconds\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2dfe6b23-cecb-4a29-9267-c45a1c0670e9",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "955c75d2-aeb1-4c2d-8de7-bfe80294c926",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "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.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}