jupyter/notebook/pyspark-nlp.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Multi-Class Text Classification Using PySpark, MLlib & Doc2Vec\n",
    "\n",
    "[Reference](https://medium.com/towards-artificial-intelligence/multi-class-text-classification-using-pyspark-mllib-doc2vec-dbfcee5b39f2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "MAX_MEMORY = \"4g\"\n",
    "#     config(\"spark.driver.memory\", MAX_MEMORY).\n",
    "\n",
    "spark = SparkSession.builder. \\\n",
    "    appName(\"pyspark-nlp\"). \\\n",
    "    config(\"spark.executor.memory\", MAX_MEMORY). \\\n",
    "    getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dataset\n",
    "https://www.kaggle.com/rmisra/news-category-dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 20 ms, sys: 40 ms, total: 60 ms\n",
      "Wall time: 22.7 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Load and repartition data\n",
    "CORES = 8\n",
    "df = spark.read.json('/dataset/news/huffingtonpost-news.json.gz').repartition(CORES * 8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "80"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|distinct_categories|\n",
      "+-------------------+\n",
      "|                 39|\n",
      "+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# df.count()\n",
    "from pyspark.sql import functions as f\n",
    "\n",
    "df.agg(f.approx_count_distinct(df.category).alias('distinct_categories')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(category='POLITICS', count=32739),\n",
       " Row(category='WELLNESS', count=17827),\n",
       " Row(category='ENTERTAINMENT', count=16058),\n",
       " Row(category='TRAVEL', count=9887),\n",
       " Row(category='STYLE & BEAUTY', count=9649),\n",
       " Row(category='PARENTING', count=8677),\n",
       " Row(category='HEALTHY LIVING', count=6694),\n",
       " Row(category='QUEER VOICES', count=6314),\n",
       " Row(category='FOOD & DRINK', count=6226),\n",
       " Row(category='BUSINESS', count=5937),\n",
       " Row(category='COMEDY', count=5175),\n",
       " Row(category='SPORTS', count=4884),\n",
       " Row(category='BLACK VOICES', count=4528),\n",
       " Row(category='HOME & LIVING', count=4195),\n",
       " Row(category='PARENTS', count=3955),\n",
       " Row(category='THE WORLDPOST', count=3664),\n",
       " Row(category='WEDDINGS', count=3651),\n",
       " Row(category='WOMEN', count=3490),\n",
       " Row(category='IMPACT', count=3459),\n",
       " Row(category='DIVORCE', count=3426),\n",
       " Row(category='CRIME', count=3405),\n",
       " Row(category='MEDIA', count=2815),\n",
       " Row(category='WEIRD NEWS', count=2670),\n",
       " Row(category='GREEN', count=2622),\n",
       " Row(category='WORLDPOST', count=2579),\n",
       " Row(category='RELIGION', count=2556),\n",
       " Row(category='STYLE', count=2254),\n",
       " Row(category='SCIENCE', count=2178),\n",
       " Row(category='WORLD NEWS', count=2177),\n",
       " Row(category='TASTE', count=2096),\n",
       " Row(category='TECH', count=2082),\n",
       " Row(category='MONEY', count=1707),\n",
       " Row(category='ARTS', count=1509),\n",
       " Row(category='FIFTY', count=1401),\n",
       " Row(category='GOOD NEWS', count=1398),\n",
       " Row(category='ARTS & CULTURE', count=1339),\n",
       " Row(category='ENVIRONMENT', count=1323),\n",
       " Row(category='COLLEGE', count=1144),\n",
       " Row(category='LATINO VOICES', count=1129),\n",
       " Row(category='CULTURE & ARTS', count=1030),\n",
       " Row(category='EDUCATION', count=1004)]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy('category').count().orderBy('count', ascending=False).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# df[df.category == 'QUEER VOICES'].select('headline').show(truncate=False)\n",
    "from pyspark.sql import functions as f\n",
    "\n",
    "CATEGORIES = {\n",
    "    'ENVIRONMENT' : ['ENVIRONMENT', 'GREEN'],\n",
    "    'POLITICS' : ['POLITICS', 'BUSINESS', 'MONEY'],\n",
    "    'TECH' : ['SCIENCE', 'TECH'],\n",
    "    'OTHERS' : ['ARTS', 'ARTS & CULTURE', 'CULTURE & ARTS', 'STYLE & BEAUTY', 'STYLE', 'TASTE', 'FOOD & DRINK', 'TRAVEL', 'PARENTING', 'DIVORCE', 'EDUCATION', 'PARENTS', 'FIFTY', 'COLLEGE', 'WEDDINGS', 'WELLNESS', 'HEALTHY LIVING', 'HOME & LIVING', 'ENTERTAINMENT', 'COMEDY', 'THE WORLDPOST', 'WORLDPOST', 'MEDIA', 'WORLD NEWS', 'IMPACT', 'WEIRD NEWS', 'GOOD NEWS', 'WOMEN', 'QUEER VOICES', 'BLACK VOICES', 'LATINO VOICES']\n",
    "}\n",
    "\n",
    "\n",
    "cats_assigned = {}\n",
    "for new_category, old_categories in CATEGORIES.items():\n",
    "    for old_category in old_categories:\n",
    "        cats_assigned[old_category] = new_category\n",
    "\n",
    "# UDF\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import StringType\n",
    "\n",
    "def reassign_categories(cat):\n",
    "    if cat in cats_assigned.keys():\n",
    "        return cats_assigned[cat]\n",
    "    else:\n",
    "        return cat\n",
    "\n",
    "reassign_categories_udf = udf(lambda cat: reassign_categories(cat), StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('new_cat', reassign_categories_udf(df.category))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------+\n",
      "|    new_cat| count|\n",
      "+-----------+------+\n",
      "|     OTHERS|141420|\n",
      "|   POLITICS| 40383|\n",
      "|     SPORTS|  4884|\n",
      "|       TECH|  4260|\n",
      "|ENVIRONMENT|  3945|\n",
      "|      CRIME|  3405|\n",
      "|   RELIGION|  2556|\n",
      "+-----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('new_cat').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reducing # of categories\n",
    "# \n",
    "# KEEP_CATS = ['POLITICS', 'TECH', 'SPORTS', 'CRIME', 'RELIGION']\n",
    "# df = df.filter(f.col('new_cat').isin(KEEP_CATS))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import functions as f\n",
    "\n",
    "df = df. \\\n",
    "    drop('category'). \\\n",
    "    drop('authors'). \\\n",
    "    drop('link'). \\\n",
    "    drop('date'). \\\n",
    "    withColumn('raw_text', f.concat(df.headline, f.lit('. '), df.short_description)). \\\n",
    "    drop('headline'). \\\n",
    "    drop('short_description'). \\\n",
    "    withColumnRenamed('new_cat', 'category'). \\\n",
    "    cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- category: string (nullable = true)\n",
      " |-- raw_text: string (nullable = true)\n",
      "\n",
      "+-----------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|category   |raw_text                                                                                                                                                                                                                                                                                                                        |\n",
      "+-----------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|OTHERS     |Easily Clean Up Easter Grass By Using Common Household Items (PHOTOS). And you're left with all that stubborn grass.                                                                                                                                                                                                            |\n",
      "|OTHERS     |Save Your Life: Get ICE. ICE stands for \"in case of emergency.\" All hospital, police, firemen and paramedics know what ICE means.  If something happens to you or someone you love, emergency personnel will search your phone for a contact called ICE.                                                                        |\n",
      "|POLITICS   |Trump Miraculously Doesn't Come Up At Paul Ryan's Farce Q&A Session. Amazing how that happened, huh?                                                                                                                                                                                                                            |\n",
      "|POLITICS   |News Roundup for July 20, 2017. News is for suckers, so suck it up. 1. Senator John McCain has been diagnosed with brain cancer. More here. 2. Trump says                                                                                                                                                                       |\n",
      "|OTHERS     |Seth Meyers Roasts Donald Trump For Not 'Winning' Like He Promised. Or, as some people may call it, losing.                                                                                                                                                                                                                     |\n",
      "|OTHERS     |New Children's Book Honors Ballet Pioneer Raven Wilkinson. Here's to the woman who paved the way for Misty Copeland.                                                                                                                                                                                                            |\n",
      "|POLITICS   |Education Department Nears Deal With Student Loan Giant Over Cheating Troops.                                                                                                                                                                                                                                                   |\n",
      "|OTHERS     |Talking to Your Child About the Dark Knight Shooting in Colorado. Stick to the facts. When they do raise it, you can ask, \"What do you want to know about the Dark Knight shooting in Colorado                                                                                                                                  |\n",
      "|ENVIRONMENT|15 Things We Wish We Could Strap A GoPro To. Shakira's Hips McKayla Maroney Hunter S. Thompson Miley’s Wrecking Ball Michael Jordan, Circa 1987 This Guy Blue Ivy Carter                                                                                                                                                        |\n",
      "|POLITICS   |Trump Faces Merkel's Art Of The Deal. By Daniel Hamilton and Niklas Helwig Two different world views will be on display this week when German Chancellor Angela                                                                                                                                                                 |\n",
      "|OTHERS     |Russia's Investigation Of Opposition Murders Does Not Bode Well For Nemtsov Case.                                                                                                                                                                                                                                               |\n",
      "|OTHERS     |This Incredible Video Shows You How To Cover Up Acne. The key to success: When using powder to mattify and set your makeup, it's crucial to apply with a dabbing motion rather                                                                                                                                                  |\n",
      "|OTHERS     |Locally Made And Globally Sourced, Cienne Found The Perfect Business Model. This fashion label has got it figured out.                                                                                                                                                                                                          |\n",
      "|OTHERS     |Men With Psoriasis Could Be More Likely To Have This Sexual Problem. No one should have to suffer in silence.                                                                                                                                                                                                                   |\n",
      "|OTHERS     |Japan FINALLY Bans Child Porn (But Not All Of It).                                                                                                                                                                                                                                                                              |\n",
      "|OTHERS     |Olympic Hammer Thrower Amanda Bingson Is A Body Positivity Queen. \"It’s a lot harder than you realize to throw naked.\"                                                                                                                                                                                                          |\n",
      "|OTHERS     |Alternative Paths to Wellness a Political Act With Winners, Losers. Who benefits when governments appeal to UNESCO to endorse a traditional medicine as intangible cultural heritage? Who loses and who gains when the FDA determines what can and cannot be called a \"medicine\"?                                               |\n",
      "|POLITICS   |Walmart Is Always Watching Its Workers. Walmart considered the Organization United for Respect at Walmart, a group that asked for more full-time jobs with higher                                                                                                                                                               |\n",
      "|ENVIRONMENT|Sinking Island Passes Fundraising Goal After Trump Says It's Not Sinking. Chesapeake Bay residents hope to build a sea wall to hold off rising waters.                                                                                                                                                                          |\n",
      "|OTHERS     |I'm Not Just Raising Little Boys, I'm Trying to Raise Good Men. Your daughters believe you when you say that there are no boys out there good enough for them. Sure, a woman doesn't need a man to survive, but let's not raise her thinking that not only does she not need them, but that there actually aren't any good ones.|\n",
      "+-----------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "import gensim.parsing.preprocessing as gsp\n",
    "from gensim import utils\n",
    "\n",
    "filters = [\n",
    "           gsp.strip_tags, \n",
    "           gsp.strip_punctuation,\n",
    "           gsp.strip_multiple_whitespaces,\n",
    "           gsp.strip_numeric,\n",
    "           gsp.remove_stopwords, \n",
    "           gsp.strip_short, \n",
    "           gsp.stem_text\n",
    "          ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Preprocessing UDF\n",
    "# https://changhsinlee.com/pyspark-udf/\n",
    "\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import StringType\n",
    "\n",
    "def clean_text(x):\n",
    "    s = x.lower()\n",
    "    s = utils.to_unicode(s)\n",
    "    for f in filters:\n",
    "        s = f(s)\n",
    "    return s\n",
    "\n",
    "clean_text_udf = udf(lambda text: clean_text(text), StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating Field to Split Dataset\n",
    "SEED=42\n",
    "TEST_SIZE=0.3\n",
    "\n",
    "from pyspark.sql.functions import rand, when\n",
    "df = df.withColumn('train', when(rand(seed=SEED) >= TEST_SIZE, True).otherwise(False))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw = df.withColumn('text', clean_text_udf(df.raw_text)).drop('raw_text')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+----------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|category   |train|text                                                                                                                                                |\n",
      "+-----------+-----+----------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|OTHERS     |true |easili clean easter grass common household item photo left stubborn grass                                                                           |\n",
      "|OTHERS     |true |save life ic ic stand case emerg hospit polic firemen paramed know ic mean happen love emerg personnel search phone contact call ic                 |\n",
      "|POLITICS   |true |trump miracul come paul ryan farc session amaz happen huh                                                                                           |\n",
      "|POLITICS   |true |new roundup juli new sucker suck senat john mccain diagnos brain cancer trump sai                                                                   |\n",
      "|OTHERS     |true |seth meyer roast donald trump win like promis peopl lose                                                                                            |\n",
      "|OTHERS     |true |new children book honor ballet pioneer raven wilkinson woman pave wai misti copeland                                                                |\n",
      "|POLITICS   |false|educ depart near deal student loan giant cheat troop                                                                                                |\n",
      "|OTHERS     |false|talk child dark knight shoot colorado stick fact rais ask want know dark knight shoot colorado                                                      |\n",
      "|ENVIRONMENT|true |thing wish strap gopro shakira hip mckayla maronei hunter thompson miley’ wreck ball michael jordan circa gui blue ivi carter                       |\n",
      "|POLITICS   |true |trump face merkel art deal daniel hamilton nikla helwig differ world view displai week german chancellor angela                                     |\n",
      "|OTHERS     |false|russia investig opposit murder bode nemtsov case                                                                                                    |\n",
      "|OTHERS     |false|incred video show cover acn kei success powder mattifi set makeup crucial appli dab motion                                                          |\n",
      "|OTHERS     |true |local global sourc cienn perfect busi model fashion label got figur                                                                                 |\n",
      "|OTHERS     |true |men psoriasi like sexual problem suffer silenc                                                                                                      |\n",
      "|OTHERS     |true |japan final ban child porn                                                                                                                          |\n",
      "|OTHERS     |true |olymp hammer thrower amanda bingson bodi posit queen it’ lot harder realiz throw nake                                                               |\n",
      "|OTHERS     |true |altern path well polit act winner loser benefit govern appeal unesco endors tradit medicin intang cultur heritag lose gain fda determin call medicin|\n",
      "|POLITICS   |true |walmart watch worker walmart consid organ unit respect walmart group ask time job higher                                                            |\n",
      "|ENVIRONMENT|false|sink island pass fundrais goal trump sai sink chesapeak bai resid hope build sea wall hold rise water                                               |\n",
      "|OTHERS     |false|rais littl boi try rais good men daughter believ boi good sure woman need man surviv let rais think need actual aren good on                        |\n",
      "+-----------+-----+----------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_raw.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------+\n",
      "|train| count|\n",
      "+-----+------+\n",
      "| true|140500|\n",
      "|false| 60353|\n",
      "+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Dataset distribution\n",
    "df_raw.groupBy('train').count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Preparing `category` field with `StringIndexer`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import StringIndexer\n",
    "str_indexer = StringIndexer(inputCol=\"category\", outputCol=\"label\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Preparing `text` field with `Word2Vec`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import Word2Vec\n",
    "from pyspark.ml.feature import Tokenizer\n",
    "\n",
    "tokenizer = Tokenizer(inputCol=\"text\", outputCol=\"tokens\")\n",
    "word2vec = Word2Vec(vectorSize=100, minCount=1, inputCol=\"tokens\", outputCol=\"features\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 220 ms, sys: 50 ms, total: 270 ms\n",
      "Wall time: 3min 22s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "from pyspark.ml import Pipeline\n",
    "\n",
    "pipeline = Pipeline(stages=[str_indexer, tokenizer, word2vec])\n",
    "model = pipeline.fit(df_raw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 100 ms, sys: 30 ms, total: 130 ms\n",
      "Wall time: 4.97 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df_w2v = model.transform(df_raw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+--------------------+-----+--------------------+--------------------+\n",
      "|   category|train|                text|label|              tokens|            features|\n",
      "+-----------+-----+--------------------+-----+--------------------+--------------------+\n",
      "|     OTHERS| true|easili clean east...|  0.0|[easili, clean, e...|[0.06482430809939...|\n",
      "|     OTHERS| true|save life ic ic s...|  0.0|[save, life, ic, ...|[0.01097476915658...|\n",
      "|   POLITICS| true|trump miracul com...|  1.0|[trump, miracul, ...|[-0.0900629305513...|\n",
      "|   POLITICS| true|new roundup juli ...|  1.0|[new, roundup, ju...|[0.10023661529911...|\n",
      "|     OTHERS| true|seth meyer roast ...|  0.0|[seth, meyer, roa...|[-0.2318754609674...|\n",
      "|     OTHERS| true|new children book...|  0.0|[new, children, b...|[0.09767392983373...|\n",
      "|   POLITICS|false|educ depart near ...|  1.0|[educ, depart, ne...|[-0.0189133385817...|\n",
      "|     OTHERS|false|talk child dark k...|  0.0|[talk, child, dar...|[-0.0075337088637...|\n",
      "|ENVIRONMENT| true|thing wish strap ...|  4.0|[thing, wish, str...|[0.02291430105397...|\n",
      "|   POLITICS| true|trump face merkel...|  1.0|[trump, face, mer...|[-0.0490362471490...|\n",
      "|     OTHERS|false|russia investig o...|  0.0|[russia, investig...|[5.85708101945264...|\n",
      "|     OTHERS|false|incred video show...|  0.0|[incred, video, s...|[0.03448039425226...|\n",
      "|     OTHERS| true|local global sour...|  0.0|[local, global, s...|[0.01297663712334...|\n",
      "|     OTHERS| true|men psoriasi like...|  0.0|[men, psoriasi, l...|[0.00233775018049...|\n",
      "|     OTHERS| true|japan final ban c...|  0.0|[japan, final, ba...|[-0.0449892252683...|\n",
      "|     OTHERS| true|olymp hammer thro...|  0.0|[olymp, hammer, t...|[0.03051877630475...|\n",
      "|     OTHERS| true|altern path well ...|  0.0|[altern, path, we...|[0.02859464146034...|\n",
      "|   POLITICS| true|walmart watch wor...|  1.0|[walmart, watch, ...|[0.01454742277772...|\n",
      "|ENVIRONMENT|false|sink island pass ...|  4.0|[sink, island, pa...|[-0.0054076819732...|\n",
      "|     OTHERS|false|rais littl boi tr...|  0.0|[rais, littl, boi...|[-0.0039671148018...|\n",
      "+-----------+-----+--------------------+-----+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "CPU times: user 0 ns, sys: 10 ms, total: 10 ms\n",
      "Wall time: 4.9 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df_w2v.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Split Datasets\n",
    "# Thanks Pedro Ferrari for this trick ;)\n",
    "import pyspark.sql.functions as f\n",
    "df_train = df_w2v.filter(f.col('train') == True)\n",
    "df_test = df_w2v.filter(f.col('train') == False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Random Forest"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "\n",
    "rf_classifier = RandomForestClassifier(labelCol=\"label\", featuresCol=\"features\")\n",
    "rf_classifier_pipeline = Pipeline(stages=[rf_classifier])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 110 ms, sys: 110 ms, total: 220 ms\n",
      "Wall time: 2min 6s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "rf_model = rf_classifier_pipeline.fit(df_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "# %%time\n",
    "# \n",
    "# rf_model.save('/dataset/news/random_forest.model')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 180 ms, sys: 100 ms, total: 280 ms\n",
      "Wall time: 1.01 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "rf_predictions = rf_model.transform(df_test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+----------+\n",
      "|   category|label|prediction|\n",
      "+-----------+-----+----------+\n",
      "|   POLITICS|  1.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|ENVIRONMENT|  4.0|       1.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|       TECH|  3.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|   POLITICS|  1.0|       1.0|\n",
      "|       TECH|  3.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     OTHERS|  0.0|       0.0|\n",
      "|     SPORTS|  2.0|       0.0|\n",
      "|     SPORTS|  2.0|       0.0|\n",
      "+-----------+-----+----------+\n",
      "only showing top 20 rows\n",
      "\n",
      "CPU times: user 50 ms, sys: 0 ns, total: 50 ms\n",
      "Wall time: 968 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "rf_predictions.select('category', 'label', 'prediction').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy = 0.782049\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "\n",
    "rf_model_evaluator = MulticlassClassificationEvaluator(\n",
    "    labelCol=\"label\", predictionCol=\"prediction\", metricName=\"accuracy\")\n",
    "\n",
    "accuracy = rf_model_evaluator.evaluate(rf_predictions)\n",
    "print(\"Accuracy = %g\" % (accuracy))"
   ]
  }
 ],
 "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}