01_fake_reviews_prep.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Fake Reviews\n",
    "\n",
    "Data Source\n",
    "\n",
    "https://www.kaggle.com/yelp-dataset/yelp-dataset\n",
    "\n",
    "The overall goal of this notebook is to create examples that we can use to train a neural network that will be able to create fake reviews."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import findspark osx\n",
    "#findspark.init()\n",
    "import pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a spark session with Apache Arrow enabled."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder. \\\n",
    "    appName('big-data') \\\n",
    "    .master(\"local[8]\") \\\n",
    "    .config(\"spark.driver.memory\", \"8g\") \\\n",
    "    .config('spark.sql.execution.arrow.enabled', 'true').getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reviews\n",
    "\n",
    "Loading the yelp reviews, printing their schema and dropping NA rows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "reviews_df = spark.read.json('yelp-dataset/yelp_academic_dataset_review.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- business_id: string (nullable = true)\n",
      " |-- cool: long (nullable = true)\n",
      " |-- date: string (nullable = true)\n",
      " |-- funny: long (nullable = true)\n",
      " |-- review_id: string (nullable = true)\n",
      " |-- stars: double (nullable = true)\n",
      " |-- text: string (nullable = true)\n",
      " |-- useful: long (nullable = true)\n",
      " |-- user_id: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "reviews_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6685900"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "reviews_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "reviews_df = reviews_df.na.drop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6685900"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "reviews_df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Businesses"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "business_df = spark.read.json('yelp-dataset/yelp_academic_dataset_business.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "192609"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "business_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- address: string (nullable = true)\n",
      " |-- attributes: struct (nullable = true)\n",
      " |    |-- AcceptsInsurance: string (nullable = true)\n",
      " |    |-- AgesAllowed: string (nullable = true)\n",
      " |    |-- Alcohol: string (nullable = true)\n",
      " |    |-- Ambience: string (nullable = true)\n",
      " |    |-- BYOB: string (nullable = true)\n",
      " |    |-- BYOBCorkage: string (nullable = true)\n",
      " |    |-- BestNights: string (nullable = true)\n",
      " |    |-- BikeParking: string (nullable = true)\n",
      " |    |-- BusinessAcceptsBitcoin: string (nullable = true)\n",
      " |    |-- BusinessAcceptsCreditCards: string (nullable = true)\n",
      " |    |-- BusinessParking: string (nullable = true)\n",
      " |    |-- ByAppointmentOnly: string (nullable = true)\n",
      " |    |-- Caters: string (nullable = true)\n",
      " |    |-- CoatCheck: string (nullable = true)\n",
      " |    |-- Corkage: string (nullable = true)\n",
      " |    |-- DietaryRestrictions: string (nullable = true)\n",
      " |    |-- DogsAllowed: string (nullable = true)\n",
      " |    |-- DriveThru: string (nullable = true)\n",
      " |    |-- GoodForDancing: string (nullable = true)\n",
      " |    |-- GoodForKids: string (nullable = true)\n",
      " |    |-- GoodForMeal: string (nullable = true)\n",
      " |    |-- HairSpecializesIn: string (nullable = true)\n",
      " |    |-- HappyHour: string (nullable = true)\n",
      " |    |-- HasTV: string (nullable = true)\n",
      " |    |-- Music: string (nullable = true)\n",
      " |    |-- NoiseLevel: string (nullable = true)\n",
      " |    |-- Open24Hours: string (nullable = true)\n",
      " |    |-- OutdoorSeating: string (nullable = true)\n",
      " |    |-- RestaurantsAttire: string (nullable = true)\n",
      " |    |-- RestaurantsCounterService: string (nullable = true)\n",
      " |    |-- RestaurantsDelivery: string (nullable = true)\n",
      " |    |-- RestaurantsGoodForGroups: string (nullable = true)\n",
      " |    |-- RestaurantsPriceRange2: string (nullable = true)\n",
      " |    |-- RestaurantsReservations: string (nullable = true)\n",
      " |    |-- RestaurantsTableService: string (nullable = true)\n",
      " |    |-- RestaurantsTakeOut: string (nullable = true)\n",
      " |    |-- Smoking: string (nullable = true)\n",
      " |    |-- WheelchairAccessible: string (nullable = true)\n",
      " |    |-- WiFi: string (nullable = true)\n",
      " |-- business_id: string (nullable = true)\n",
      " |-- categories: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- hours: struct (nullable = true)\n",
      " |    |-- Friday: string (nullable = true)\n",
      " |    |-- Monday: string (nullable = true)\n",
      " |    |-- Saturday: string (nullable = true)\n",
      " |    |-- Sunday: string (nullable = true)\n",
      " |    |-- Thursday: string (nullable = true)\n",
      " |    |-- Tuesday: string (nullable = true)\n",
      " |    |-- Wednesday: string (nullable = true)\n",
      " |-- is_open: long (nullable = true)\n",
      " |-- latitude: double (nullable = true)\n",
      " |-- longitude: double (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- postal_code: string (nullable = true)\n",
      " |-- review_count: long (nullable = true)\n",
      " |-- stars: double (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "business_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "business_df = business_df.na.drop(subset=['city', 'categories'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "192127"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "business_df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We ultimately want to create examples in the following form:\n",
    "\n",
    "'RATING CITY CATEGORIES': 'REVIEW TEXT'\n",
    "\n",
    "If we wanted more granular control over our fake review generation, we could also use something like 'RATING BUSINESS_NAME CITY STATE CATEGORIES'\n",
    "\n",
    "Let's keep it simple and create a new examples data frame:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "reviews = reviews_df.alias('reviews')\n",
    "business = business_df.alias('business')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import concat_ws\n",
    "\n",
    "examples = reviews.join(business, reviews.business_id == business.business_id) \\\n",
    "    .select(concat_ws(' ', reviews.stars, business.city, business.categories).alias('context'), \\\n",
    "            reviews.text.alias('review'))\n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "|             context|              review|\n",
      "+--------------------+--------------------+\n",
      "|1.0 Las Vegas Fit...|Total bill for th...|\n",
      "|5.0 Las Vegas Bea...|I *adore* Travis ...|\n",
      "|5.0 Chandler Heal...|I have to say tha...|\n",
      "|5.0 Calgary Bars,...|Went in for a lun...|\n",
      "|1.0 Scottsdale Te...|Today was my seco...|\n",
      "|4.0 Pittsburgh Re...|I'll be the first...|\n",
      "|3.0 Markham Food,...|Tracy dessert had...|\n",
      "|1.0 Scottsdale Sp...|This place has go...|\n",
      "|2.0 Cleveland Bre...|I was really look...|\n",
      "|3.0 Las Vegas Sho...|It's a giant Best...|\n",
      "|4.0 Las Vegas Per...|Like walking back...|\n",
      "|1.0 Mesa Restaura...|Walked in around ...|\n",
      "|4.0 Pittsburgh It...|Wow. So surprised...|\n",
      "|4.0 Las Vegas Hot...|Michael from Red ...|\n",
      "|1.0 Toronto Asian...|I cannot believe ...|\n",
      "|5.0 Toronto Sandw...|You can't really ...|\n",
      "|4.0 Orange Villag...|Great lunch today...|\n",
      "|3.0 Phoenix Carib...|I love chinese fo...|\n",
      "|5.0 Chandler Sand...|We've been a huge...|\n",
      "|3.0 Toronto Resta...|Good selection of...|\n",
      "+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "examples.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6683763"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "examples.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Tokenization\n",
    "\n",
    "We need to tokenize our examples so that we can feed them into our neural network. We also need to clean up our examples before tokenizing them. We remove new-lines and non-ascii characters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import regexp_replace\n",
    "examples = examples.withColumn('review', regexp_replace(examples.review, '[\\\\r\\\\n]', ' '))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "examples = examples.withColumn('review', regexp_replace(examples.review, '[^\\x00-\\x7F]+', ' '))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We now want to save the context and review columns as text files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "examples.select(examples.context).write.format('text').save('context.txt')\n",
    "examples.select(examples.review).write.format('text').save('reviews.txt')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Making sure that both exports have the same line counts"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6683763\r\n"
     ]
    }
   ],
   "source": [
    "!cat context.txt/*.txt | wc -l"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6683763\r\n"
     ]
    }
   ],
   "source": [
    "!cat reviews.txt/*.txt | wc -l"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Unfortunately, the default SparkML Tokenizer is rather simple, there is a RegexTokenizer, but crafting hand-written tokenization rules with it is error-prone and cumbersome. We could try to use an industrial-strength Tokenizer from spacy. Implementing an SparkML-Transformer would require a Java counter-part on top of the python implementation, so let's fall back to UDF's instead.\n",
    "\n",
    "Another alternative is using Apache Arrow.\n",
    "\n",
    "As it turns out, both ways really slow. The following cells are therefore only used for illustrative purposes. I've tested the UDF approach multiple times and had a lot of time outs or OutOfMemory exceptions.\n",
    "\n",
    "I therefore had to rely on Stanford's CoreNLP package to tokenize the examples. Please consult the tokenization notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import spacy\n",
    "from pyspark.sql.functions import udf, pandas_udf\n",
    "from pyspark.sql.types import ArrayType, StringType\n",
    "\n",
    "nlp = spacy.load(\"en_core_web_sm\")\n",
    "\n",
    "@udf(ArrayType(StringType()))\n",
    "def tokenize(s):\n",
    "    return [token.text for token in nlp(s)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import spacy\n",
    "from pyspark.sql.functions import udf, pandas_udf, PandasUDFType\n",
    "from pyspark.sql.types import ArrayType, StringType\n",
    "\n",
    "nlp = spacy.load(\"en_core_web_sm\")\n",
    "\n",
    "def spacy_tokenize(s):\n",
    "    return [token.text for token in nlp(s)]\n",
    "\n",
    "@pandas_udf(\"string\", PandasUDFType.SCALAR)\n",
    "def tokenize(x):\n",
    "    return x.apply(spacy_tokenize)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tokenized = examples.select(concat_ws(' ', tokenize(examples.context)).alias('context'), concat_ws(' ', tokenize(examples.review)).alias('review'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, val, test = tokenized.randomSplit([0.98, 0.016, 0.004], seed=42)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train.count()\n",
    "val.count()\n",
    "test.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def write_df(df, file_name):\n",
    "    df.select(df.context).write.format('text').save(file_name + '_src.txt')\n",
    "    df.select(df.review).write.format('text').save(file_name + '_tgt.txt')\n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train = train.cache()\n",
    "val = val.cache()\n",
    "test = test.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "write_df(test, 'test')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "write_df(val, 'val')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "write_df(train, 'train')"
   ]
  }
 ],
 "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
}