FP-Growth algorithm.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "particular-transportation",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "from pyspark import SparkContext\n",
    "import json\n",
    "import time\n",
    "import collections\n",
    "from operator import add\n",
    "from pyspark.mllib.fpm import FPGrowth"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "massive-bracket",
   "metadata": {},
   "outputs": [],
   "source": [
    "startTime = time.time()\n",
    "\n",
    "filter_threshold = 70\n",
    "global_threshold = 50\n",
    "input_file_path = r'user_business.csv'\n",
    "output_file_path = r'task3_Output.txt'\n",
    "\n",
    "sc = SparkContext.getOrCreate()\n",
    "\n",
    "def turnStr2Pair(pairStr):\n",
    "    return (pairStr.split(',')[0], pairStr.split(',')[1])\n",
    "    \n",
    "# read csv file and generate pairs\n",
    "with open(input_file_path) as f:\n",
    "    rawStrList = f.readlines()[1:]\n",
    "    pairList = [pair.split('\\n')[0] for pair in rawStrList]\n",
    "    f.close()\n",
    "\n",
    "qualifiedUsersRDD = sc.parallelize(pairList).map(lambda pairStr:turnStr2Pair(pairStr))\\\n",
    "                    .groupByKey().mapValues(lambda iterable:set(iterable))\\\n",
    "                    .filter(lambda user_bus_pair:len(user_bus_pair[1])>filter_threshold)\\\n",
    "                    .map(lambda pair: pair[1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "increasing-plumbing",
   "metadata": {},
   "outputs": [],
   "source": [
    "minSupport = global_threshold / qualifiedUsersRDD.count()\n",
    "fpgRDD = FPGrowth.train(qualifiedUsersRDD, minSupport, 50)\n",
    "result = sorted(fpgRDD.freqItemsets().collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "coordinated-tyler",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Length of task2 result: 2582\n"
     ]
    }
   ],
   "source": [
    "task2RawResultList = list()\n",
    "task2ResultList = list()\n",
    "fileReader = open('task2Output.txt','r')\n",
    "foundFrequentItemsetsSection = False\n",
    "frequentItemsetsString = ''\n",
    "for line in fileReader:\n",
    "    if not foundFrequentItemsetsSection:\n",
    "        if line=='Frequent Itemsets: \\n':\n",
    "            foundFrequentItemsetsSection = True\n",
    "    else:\n",
    "        frequentItemsetsString+=line\n",
    "task2RawResultList = frequentItemsetsString.replace('\\n\\n',',').split('),(')\n",
    "\n",
    "for frequentItemset in task2RawResultList:\n",
    "    cleanItem = frequentItemset.replace('\"','').replace('(','').replace(')','').replace('\\'','')\n",
    "    cleanItemList = cleanItem.split(', ')\n",
    "    task2ResultList.append(tuple(cleanItemList))\n",
    "print('Length of task2 result:', len(task2ResultList))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "introductory-blade",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Length of task3 result: 2582\n"
     ]
    }
   ],
   "source": [
    "fpgResultList = list()\n",
    "for frequentItem in result:\n",
    "    fpgResultList.append(tuple(frequentItem[0]))\n",
    "fpgResultList.sort(key=lambda l:(len(l),l))\n",
    "print('Length of task3 result:', len(fpgResultList))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "existing-wagon",
   "metadata": {},
   "outputs": [],
   "source": [
    "intersection = set(task2ResultList).intersection(set(fpgResultList))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "therapeutic-gazette",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Duration: 88\n"
     ]
    }
   ],
   "source": [
    "outputFile = open(output_file_path, 'w')\n",
    "outputFile.write('Task2,'+str(len(task2ResultList))+'\\n')\n",
    "outputFile.write('Task3,'+str(len(fpgResultList))+'\\n')\n",
    "outputFile.write('Intersection,'+str(len(intersection))+'\\n')\n",
    "outputFile.close()\n",
    "\n",
    "print(\"Duration: %d\" % (time.time() - startTime))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}