B_prediction_using_spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Prediction using PySpark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sc not defined\n"
     ]
    }
   ],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "from pyspark import SparkConf, SparkContext\n",
    "\n",
    "try:\n",
    "    sc.stop()\n",
    "except NameError:\n",
    "    print(\"sc not defined\")\n",
    "\n",
    "config = SparkConf().setMaster(\"local[*]\").setAppName(\"ClassifyUrl\")    \n",
    "sc = SparkContext(conf=config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "                    .master(\"local\") \\\n",
    "                    .appName(\"Classify Urls\") \\\n",
    "                    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_csv = 'bigdata/train.csv'\n",
    "html_csv = 'bigdata/train/html_data.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Webpage_id: integer (nullable = true)\n",
      " |-- Domain: string (nullable = true)\n",
      " |-- Url: string (nullable = true)\n",
      " |-- Tag: string (nullable = true)\n",
      "\n",
      "Wall time: 7.2 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "train = spark.read.csv(train_csv, \n",
    "                       header=True,\n",
    "                      inferSchema=True)\n",
    "\n",
    "# Analyze Data types in dataset\n",
    "train.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total Records in Training Dataset : 53447\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(Webpage_id=1, Domain='www.fiercepharma.com', Url='http://www.fiercepharma.com/marketing/tecfidera-gilenya-and-aubagio-s-3-way-battle-for-ms-share-about-to-get-more-interesting', Tag='news'),\n",
       " Row(Webpage_id=2, Domain='www.fiercepharma.com', Url='http://www.fiercepharma.com/pharma/novo-equipped-to-weather-storm-u-s-diabetes-market-ceo-says', Tag='news'),\n",
       " Row(Webpage_id=3, Domain='www.fiercepharma.com', Url='http://www.fiercepharma.com/pharma/another-exec-departs-troubled-endo-and-time-it-s-for-another-drugmaker', Tag='news'),\n",
       " Row(Webpage_id=4, Domain='www.fiercepharma.com', Url='http://www.fiercepharma.com/pharma/teva-buy-biosim-specialist-celltrion-it-wouldn-t-say-no', Tag='news'),\n",
       " Row(Webpage_id=5, Domain='www.fiercepharma.com', Url='http://www.fiercepharma.com/marketing/actress-marissa-tomei-partners-allergan-restasis-to-drive-dry-eye-awareness', Tag='news')]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "print('Total Records in Training Dataset :', train.count())\n",
    "train.head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+--------------------+----+\n",
      "|Webpage_id|              Domain|                 Url| Tag|\n",
      "+----------+--------------------+--------------------+----+\n",
      "|         1|www.fiercepharma.com|http://www.fierce...|news|\n",
      "|         2|www.fiercepharma.com|http://www.fierce...|news|\n",
      "|         3|www.fiercepharma.com|http://www.fierce...|news|\n",
      "|         4|www.fiercepharma.com|http://www.fierce...|news|\n",
      "|         5|www.fiercepharma.com|http://www.fierce...|news|\n",
      "+----------+--------------------+--------------------+----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# head() gives Ugly Output :(\n",
    "# Prefer show() over head()\n",
    "train.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No. of cols in train dataset :  4\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "['Webpage_id', 'Domain', 'Url', 'Tag']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# How many columns do we have in train and what are their names?\n",
    "print('No. of cols in train dataset : ', len(train.columns))\n",
    "train.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+--------------------+--------------------+--------------+\n",
      "|summary|        Webpage_id|              Domain|                 Url|           Tag|\n",
      "+-------+------------------+--------------------+--------------------+--------------+\n",
      "|  count|             53447|               53447|               53447|         53447|\n",
      "|   mean| 39920.78603102139|                null|                null|          null|\n",
      "| stddev|22945.942450142324|                null|                null|          null|\n",
      "|    min|                 1|  1.eyefortravel.com|http://1.eyefortr...|clinicalTrials|\n",
      "|    max|             79345|zoonosis.conferen...|https://zoologica...|        thesis|\n",
      "+-------+------------------+--------------------+--------------------+--------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'\\nObservation:\\nAs we can see that, describe operation is working for String type column but the output for mean, stddev are null and min & max values are calculated based on ASCII value of categories.\\n'"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# How to get the summary statistics (mean, standard deviance, min ,max , count) of numerical columns in a DataFrame?\n",
    "train.describe().show()\n",
    "\n",
    "'''\n",
    "Observation:\n",
    "As we can see that, describe operation is working for String type column but the output for mean, stddev are null and min & max values are calculated based on ASCII value of categories.\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+----+\n",
      "|Webpage_id|              Domain| Tag|\n",
      "+----------+--------------------+----+\n",
      "|         1|www.fiercepharma.com|news|\n",
      "|         2|www.fiercepharma.com|news|\n",
      "|         3|www.fiercepharma.com|news|\n",
      "|         4|www.fiercepharma.com|news|\n",
      "|         5|www.fiercepharma.com|news|\n",
      "+----------+--------------------+----+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 400 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# How to select column(s) from the DataFrame?\n",
    "train.select('Webpage_id','Domain','Tag').show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Wall time: 10.9 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "(3974, 9)"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "# How to find the number of distinct Domain and Tags (Target-Classes) in train files?\n",
    "train.select('Domain').distinct().count(), train.select('Tag').distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Count of Missing values in Train Dataset :\n",
      "\n",
      "Domain : 0\n",
      "Url : 0\n",
      "Tag : 0\n"
     ]
    }
   ],
   "source": [
    "# Check for Null values in \n",
    "print('Count of Missing values in Train Dataset :\\n')\n",
    "print('Domain :', train.filter(train.Domain.isNull()).count())\n",
    "print('Url :', train.filter(train.Url.isNull()).count())\n",
    "print('Tag :', train.filter(train.Tag.isNull()).count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "53447"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# How to drop the all rows with null value?\n",
    "train.dropna().count() # Count of rows in newly returned non-null dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "53447"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# How to fill the null values in Domain column of DataFrame with, some constant value, say, 'www.missing.in'?\n",
    "missing_domain = 'www.missing.in'\n",
    "train.fillna(missing_domain, 'Domain').count() # Count of rows in newly returned non-null dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+--------------------+-----------+\n",
      "|Webpage_id|              Domain|                 Url|        Tag|\n",
      "+----------+--------------------+--------------------+-----------+\n",
      "|     18472|cancerci.biomedce...|https://cancerci....|publication|\n",
      "|     26364|www.naturalnews.c...|http://www.natura...|       news|\n",
      "|     28191|         twitter.com|https://twitter.c...|     others|\n",
      "+----------+--------------------+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# How to create a sample DataFrame from the base DataFrame?\n",
    "train.sample(False, # withReplacement=False\n",
    "             0.0001, # fraction = x percecntage that we want to pick\n",
    "             42 # seed to reproduce the result\n",
    "            ).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|              Domain|count|\n",
      "+--------------------+-----+\n",
      "|thesis.library.ca...|  301|\n",
      "|academiccommons.c...|  300|\n",
      "|  www.dart-europe.eu|  300|\n",
      "|       curate.nd.edu|  300|\n",
      "|      dspace.mit.edu|  300|\n",
      "|ecommons.cornell.edu|  300|\n",
      "|     www.nice.org.uk|  230|\n",
      "|www.ncbi.nlm.nih.gov|  226|\n",
      "+--------------------+-----+\n",
      "\n",
      "+--------------------+-----+\n",
      "|              Domain|count|\n",
      "+--------------------+-----+\n",
      "|thesis.library.ca...|  301|\n",
      "|academiccommons.c...|  300|\n",
      "|  www.dart-europe.eu|  300|\n",
      "|       curate.nd.edu|  300|\n",
      "|      dspace.mit.edu|  300|\n",
      "|ecommons.cornell.edu|  300|\n",
      "|     www.nice.org.uk|  230|\n",
      "|www.ncbi.nlm.nih.gov|  226|\n",
      "+--------------------+-----+\n",
      "\n",
      "+--------------------+-----+\n",
      "|              Domain|count|\n",
      "+--------------------+-----+\n",
      "|thesis.library.ca...|  301|\n",
      "|academiccommons.c...|  300|\n",
      "|  www.dart-europe.eu|  300|\n",
      "|       curate.nd.edu|  300|\n",
      "|      dspace.mit.edu|  300|\n",
      "|ecommons.cornell.edu|  300|\n",
      "|     www.nice.org.uk|  230|\n",
      "|www.ncbi.nlm.nih.gov|  226|\n",
      "+--------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# How to find the number of rows we have per Domain?\n",
    "from pyspark.sql.functions import col\n",
    "train.groupBy('Domain') \\\n",
    "    .count() \\\n",
    "    .filter(\"`count` > 225\") \\\n",
    "    .sort(col('count').desc()) \\\n",
    "    .show(10) # Show Count of Top 10\n",
    "\n",
    "# Or like below:\n",
    "from pyspark.sql.functions import desc\n",
    "train.groupBy('Domain') \\\n",
    "    .count() \\\n",
    "    .filter(\"`count` > 225\") \\\n",
    "    .sort(desc('count')) \\\n",
    "    .show(10) # Show Count of Top 10\n",
    "\n",
    "# Or like below:\n",
    "from pyspark.sql.functions import desc\n",
    "train.groupBy('Domain') \\\n",
    "    .count() \\\n",
    "    .filter(\"`count` > 225\") \\\n",
    "    .orderBy('count', ascending=False) \\\n",
    "    .show(10) # Show Count of Top 10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+--------------------+----+\n",
      "|Webpage_id|      Domain|                 Url| Tag|\n",
      "+----------+------------+--------------------+----+\n",
      "|         1|fiercepharma|http://www.fierce...|news|\n",
      "|         2|fiercepharma|http://www.fierce...|news|\n",
      "|         3|fiercepharma|http://www.fierce...|news|\n",
      "|         4|fiercepharma|http://www.fierce...|news|\n",
      "|         5|fiercepharma|http://www.fierce...|news|\n",
      "+----------+------------+--------------------+----+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 3.82 s\n"
     ]
    }
   ],
   "source": [
    "%%time \n",
    "# Wall time: 2.13s - 2.39s\n",
    "\n",
    "# OBJECTIVE : Get just the domain from URLs\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import StringType\n",
    "import tldextract\n",
    "\n",
    "def extract_domain(url):\n",
    "    return tldextract.extract(url).domain\n",
    "\n",
    "extract_domain_udf = udf(extract_domain, StringType())\n",
    "# extract_domain_udf = udf(lambda url : tldextract.extract(url).domain, StringType())\n",
    "\n",
    "train = train.withColumn('Domain', extract_domain_udf(train.Domain))\n",
    "train.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "from bs4 import BeautifulSoup # For Scraping HTML page\n",
    "from bs4.element import Comment\n",
    "import re"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Objective : Extract text from title tag of HTML source of web-page\n",
    "def extract_title(page):\n",
    "    if (page == None): \n",
    "        return None\n",
    "    soup = BeautifulSoup(page, 'html.parser')\n",
    "    title_tag = soup.find('title')\n",
    "    if (title_tag == None):\n",
    "        title = None\n",
    "    else:\n",
    "        title = title_tag.text.encode('utf-8',errors='ignore').decode('utf-8').strip()\n",
    "    return title"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "#  OBJECTIVE: Functions to parse HTML content and extract text that matters.\n",
    "def extract_body(page):\n",
    "    if (page == None): \n",
    "        return None\n",
    "    soup = BeautifulSoup(page, 'html.parser', from_encoding=\"utf-8\")\n",
    "    body_tag = soup.find('body')\n",
    "    if (body_tag == None):\n",
    "        body = page \n",
    "    else:\n",
    "        body = body_tag # What should be returned here? How to stringify this for further  procecssing?\n",
    "    return body\n",
    "\n",
    "def is_visible_content(element):\n",
    "    if element.parent.name in ['style', 'script', 'head', 'title', 'meta', '[document]']:\n",
    "        return False\n",
    "    if isinstance(element, Comment):\n",
    "        return False\n",
    "    return True\n",
    "\n",
    "def remove_extra_spaces(str):\n",
    "    return u\" \".join(str.split())\n",
    "\n",
    "def extract_text(page):\n",
    "    if (page == None): \n",
    "        return None\n",
    "    soup = BeautifulSoup(page, 'html.parser') #, from_encoding=\"utf-8\"\n",
    "    texts = soup.findAll(text=True) # Extracts text from all HTML Markups, incl nested ones\n",
    "    visible_texts = filter(is_visible_content, texts)\n",
    "    # The u-prefix u\" \".join() indicates Unicode and has been in python since v2.0\n",
    "    # Ref. Read: https://www.joelonsoftware.com/2003/10/08/the-absolute-minimum-every-software-developer-absolutely-positively-must-know-about-unicode-and-character-sets-no-excuses/\n",
    "    text = u\" \".join(remove_extra_spaces(t.strip()) for t in visible_texts)\n",
    "    text = text.replace(',','')\n",
    "    text = text.replace('|','')\n",
    "    text = re.sub(r'\\s\\s+',' ',text).strip()\n",
    "    return text.encode('utf-8',errors='ignore').decode('utf-8').strip()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'spark_parser_mode'"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "config.get(\"spark\", \"spark_parser_mode\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Webpage_id: integer (nullable = true)\n",
      " |-- Html: string (nullable = true)\n",
      "\n",
      "Wall time: 1min 44s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# OBJECTIVE : Read html_data.csv\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.sql.functions import lit # lit for literals\n",
    "\n",
    "extract_text_udf = udf(extract_text, StringType())\n",
    "\n",
    "from pyspark.sql.types import StructType, StructField\n",
    "from pyspark.sql.types import DoubleType, IntegerType, StringType\n",
    "schema = StructType([\n",
    "    StructField(\"Webpage_id\", IntegerType()),\n",
    "    StructField(\"Html\", StringType())\n",
    "])\n",
    "\n",
    "html_df = spark.read.csv(html_csv, \n",
    "                         header=True, \n",
    "                         multiLine=True, \n",
    "                         ignoreLeadingWhiteSpace=True, \n",
    "                         ignoreTrailingWhiteSpace=True, \n",
    "                         encoding=\"UTF-8\",\n",
    "                         sep=',',\n",
    "                         quote='\"', \n",
    "                         escape='\"',\n",
    "                         maxColumns=2,\n",
    "#                          mode='spark_parser_mode',\n",
    "#                          schema=schema)\n",
    "                         inferSchema=True)\n",
    "'''\n",
    "html_df = spark.read.format('csv') \\\n",
    "                    .option('header',True) \\\n",
    "                    .option('ignoreLeadingWhiteSpace',True) \\\n",
    "                    .option('ignoreTrailingWhiteSpace',True) \\\n",
    "                    .option('inferSchema',True) \\\n",
    "                    .option('maxColumns',2) \\\n",
    "                    .load(html_csv)\n",
    "'''\n",
    "# Analyze Data types in dataset\n",
    "html_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+-----+\n",
      "|Webpage_id|                Html|Title|\n",
      "+----------+--------------------+-----+\n",
      "|         1|<!DOCTYPE html>\n",
      "<...|     |\n",
      "|         2|<!DOCTYPE html>\n",
      "<...|     |\n",
      "|         3|<!DOCTYPE html>\n",
      "<...|     |\n",
      "|         4|<!DOCTYPE html>\n",
      "<...|     |\n",
      "|         5|<!DOCTYPE html>\n",
      "<...|     |\n",
      "+----------+--------------------+-----+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 2min 43s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Adding a constant column \n",
    "html_df = html_df.withColumn('Title',lit(''))\n",
    "html_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "ename": "AttributeError",
     "evalue": "'DataFrame' object has no attribute 'shape'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
      "\u001b[1;32m<timed exec>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n",
      "\u001b[1;32mC:\\Spark\\python\\pyspark\\sql\\dataframe.py\u001b[0m in \u001b[0;36m__getattr__\u001b[1;34m(self, name)\u001b[0m\n\u001b[0;32m   1180\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mname\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcolumns\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1181\u001b[0m             raise AttributeError(\n\u001b[1;32m-> 1182\u001b[1;33m                 \"'%s' object has no attribute '%s'\" % (self.__class__.__name__, name))\n\u001b[0m\u001b[0;32m   1183\u001b[0m         \u001b[0mjc\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jdf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mapply\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mname\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1184\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mColumn\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mjc\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAttributeError\u001b[0m: 'DataFrame' object has no attribute 'shape'"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# Wall time: 1h 10min 6s\n",
    "\n",
    "# OBJECTIVE : From html_data.csv loaded in df, extract title and text from html-page, and add the them to train.csv as new columns\n",
    "\n",
    "# Transforming an existing column\n",
    "html_df = html_df.withColumn('Html',extract_text_udf(html_df.Html))\n",
    "html_df = html_df.withColumnRenamed('Html','Html2Text')\n",
    "html_df.write.csv('bigdata/train/sparkoutput', header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "79345"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "html_df.count() # 79345"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(Webpage_id=10, Html2Text='Skip to main content Twitter LinkedIn Search Top Menu DDF 2017 FierceBiotech Jobs Resources Events Subscribe Main navigation Pharma M&A Regulatory Financials Corporate Legal Manufacturing M&A Outsourcing Regulatory Supply Chain Partnering Drug Safety Marketing Regulatory DTC Advertising Digital and Social Media Data and Analytics Launches Pharma Asia M&A R&D Regulatory Sales and Marketing Financials Manufacturing Animal Health R&D M&A Regulatory Veterinarian Financials Vaccines Drug Delivery R&D Regulatory Partnering Vaccines Deals Infectious Diseases R&D Regulatory Main navigation - Mobile Pharma M&A Regulatory Financials Corporate Legal Manufacturing M&A Outsourcing Regulatory Supply Chain Partnering Drug Safety Marketing Regulatory DTC Advertising Digital and Social Media Data and Analytics Launches Pharma Asia M&A R&D Regulatory Sales and Marketing Financials Manufacturing Animal Health R&D M&A Regulatory Veterinarian Financials Vaccines Drug Delivery R&D Regulatory Partnering Vaccines Deals Infectious Diseases R&D Regulatory Pharma A silver lining to the drug-pricing uproar? For generic competition yes GPhA chief says by Eric Sagonowsky Sep 20 2016 9:58am GPhA CEO Chip Davis Answering some recent news about \"extraordinary\" generic price increasesÂ Davis emphasized in a presentation onÂ Monday that those hikes were few and far between in the copycat-drug universe. If you\\'re looking for drug-cost drivers look no further thanÂ specialty drugs which in 2014 accounted for 31.8% of spending on 1% of the population said Davis who heads up the Generic Pharmaceutical Association. And specialty pharma\\'s share of drug spendingÂ is expected to hit 50% by 2018 Davis said inÂ remarks Â prepared for a call with the media. Alternatively generics make up 88% of prescriptions but account for 28% of costs he said. As a drug-pricing firestorm has grown since last year--touched off by high-profile price increases by Turing Pharmaceuticals and Valeant Pharmaceuticals--reports of big hikesÂ have gone viral and made their way into daily conversation. Among other companies Mylan hasÂ found itself in a harsh spotlight for its multiyear 400%-plus price increase on its anaphylaxis injectionÂ EpiPen. Some lawmakers have questioned why generic alternativesÂ werenâ€™t more available. To that point the FDAâ€™s Dr. Doug Throckmorton is set to appear at a hearing tomorrow of the House Committee on Oversight and Government Reform and he\\'s expected to face questions about ways to speed alternatives to market. MylanÂ CEO Heather Bresch will also testify. Davis said all of the pricingÂ outcry â€œhas the potential to bear positive resultsâ€� because it could lead to increased competition for branded meds. First though the generics industry will need to fend off efforts by some branded pharma companies to block competition. Those include lobbying to hinder biosim uptake opposing â€œa fair patent review processâ€� and a push to defeat legislation that would â€œcurb anti-competitive abuses that are being used to delay generic drug developmentâ€� Davis said. GPhAÂ wants to see those tactics come to an end. Davis\\' proclamationÂ comes less than a month after a GAO report found that generic drug prices in Medicare Part D decreased significantly in recent years as a groupÂ falling 59% from 2010 to 2015. But even still the GAO noted that some drugs saw â€œextraordinaryâ€� price increases over the period with their stickers growing by at least 100%. GPhA examined those extraordinary hikes mentioned in the GAO report. For meds that had data publicly available their average increase was 65 cents between 2013 and 2015. That meant anÂ increase from an average of $1.62 to $2.28 per unit according to figures cited by Davis. Pointing to hikes such as those by Mylan and Turing Democratic presidential candidate Hillary Clinton outlined a drug price plan earlier this month to tackle â€œexcessive outlierâ€� price increases. Sheâ€™s proposing a panel that would identify bad actors and then impose fines or other enforcement actions. Related Articles: \\'Extraordinary\\' generics price hikes hit Medicare Part D amid big reduction overall Amid still-raging EpiPen scandal Clinton rolls out plan to fight drug price hikes Mylan CEO faces grilling on Capitol Hill as senators unveil price-hike legislation Some generic drug prices soar despite heavy competition Pfizer CEO calls Clinton drug plan \\'very negative for innovation\\' Read more on generics drug prices drug spending GPhA Hillary Clinton Mylan Turing Pharmaceuticals Valeant Pharmaceuticals Goverment Accountability Office Chip Davis Popular Content Lilly to cut 3500 jobs take a $1.2B hit as it aims for $500M in savings Sep 07 2017 After Keytruda deaths FDA hits pause on Bristol-Myers Squibb\\'s Opdivo myeloma trials Sep 07 2017 Survival win has Bristol-Myers\\' Opdivo-Yervoy combo looking better as first-line kidney cancer treatment Sep 07 2017 Senatorâ€™s report documents deceit by former employees of embattled Insys Sep 07 2017 Novo to pay nearly $60M to settle Victoza marketing suits with DOJ whistleblowers Sep 06 2017 About the Author Eric Sagonowsky Associate Editor esagonowsky@fiercemarkets.com â€‹ Home Subscribe Manage Newsletter Subscriptions Advertise RSS Privacy About Us Contact Â© 2017 Questex LLC. All rights reserved. 275 Grove Street Suite 2-130 Newton MA 02466 Reproduction in whole or part is prohibited.', Title='')"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "html_df.head(10)[9]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}