LDA+Aliens.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: findspark in /Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages\r\n"
     ]
    }
   ],
   "source": [
    "#from https://community.hortonworks.com/articles/84781/spark-text-analytics-uncovering-data-driven-topics.html\n",
    "from collections import defaultdict\n",
    "from pyspark import SparkContext\n",
    "from pyspark.mllib.linalg import Vector, Vectors\n",
    "from pyspark.mllib.clustering import LDA, LDAModel\n",
    "from pyspark.sql import SQLContext\n",
    "import re\n",
    "! pip install findspark \n",
    "import findspark\n",
    "\n",
    "# Or the following command\n",
    "findspark.init(\"/usr/local/opt/apache-spark/libexec\")\n",
    "\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "\n",
    "num_of_stop_words = 100      # Number of most common words to remove, trying to eliminate stop words\n",
    "num_topics = 5\t            # Number of topics we are looking for\n",
    "num_words_per_topic = 10    # Number of words to display for each topic\n",
    "max_iterations = 20         # Max number of times to iterate before finishing\n",
    "\n",
    "# Initialize\n",
    "# spark = SparkSession \\\n",
    "#     .builder \\\n",
    "#     .appName(\"Python Spark SQL basic example\") \\\n",
    "#     .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "#     .getOrCreate()\n",
    "sc = SparkContext('local', 'PySPARK LDA Example')\n",
    "sql_context = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "os.environ[\"SPARK_HOME\"] = \"/usr/local/Cellar/apache-spark/1.5.1/\"\n",
    "os.environ[\"PYSPARK_PYTHON\"]=\"python3\"\n",
    "os.environ[\"PYSPARK_DRIVER_PYTHON\"]=\"python3\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = sc.wholeTextFiles('/Users/edwinv/Dropbox/testing/*', minPartitions=300).map(lambda x: x[1])\n",
    "\n",
    "tokens = data                                                   \\\n",
    "    .map( lambda document: document.strip().lower())            \\\n",
    "    .map( lambda document: re.split(\"[\\s;,#]\", document))       \\\n",
    "    .map( lambda word: [x for x in word if x.isalpha()])        \\\n",
    "    .map( lambda word: [x for x in word if len(x) > 3] )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "termCounts = tokens                             \\\n",
    "    .flatMap(lambda document: document)         \\\n",
    "    .map(lambda word: (word, 1))                \\\n",
    "    .reduceByKey( lambda x,y: x + y)            \\\n",
    "    .map(lambda tuple: (tuple[1], tuple[0]))    \\\n",
    "    .sortByKey(False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "threshold_value = termCounts.take(num_of_stop_words)[num_of_stop_words - 1][0]\n",
    "\n",
    "# Only keep words with a count less than the threshold identified above, \n",
    "# and then index each one and collect them into a map\n",
    "vocabulary = termCounts                         \\\n",
    "    .filter(lambda x : x[0] < threshold_value)  \\\n",
    "    .map(lambda x: x[1])                        \\\n",
    "    .zipWithIndex()                             \\\n",
    "    .collectAsMap()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "def document_vector(document):\n",
    "    id = document[1]\n",
    "    counts = defaultdict(int)\n",
    "    for token in document[0]:\n",
    "        if token in vocabulary:\n",
    "            token_id = vocabulary[token]\n",
    "            counts[token_id] += 1\n",
    "    counts = sorted(counts.items())\n",
    "    keys = [x[0] for x in counts]\n",
    "    values = [x[1] for x in counts]\n",
    "    return (id, Vectors.sparse(len(vocabulary), keys, values))\n",
    "\n",
    "# Process all of the documents into word vectors using the \n",
    "# `document_vector` function defined previously\n",
    "documents = tokens.zipWithIndex().map(document_vector).map(list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[551] at RDD at PythonRDD.scala:48"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "documents"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "inv_voc = {value: key for (key, value) in vocabulary.items()}\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open(\"corpus.txt\", 'w') as f:\n",
    "    lda_model = LDA.train(documents, k=num_topics, maxIterations=max_iterations)\n",
    "\n",
    "    topic_indices = lda_model.describeTopics(maxTermsPerTopic=num_words_per_topic)\n",
    "        \n",
    "    # Print topics, showing the top-weighted 10 terms for each topic\n",
    "    for i in range(len(topic_indices)):\n",
    "        f.write(\"Topic #{0}\\n\".format(i + 1))\n",
    "        for j in range(len(topic_indices[i][0])):\n",
    "            f.write(\"{0}\\t{1}\\n\".format(inv_voc[topic_indices[i][0][j]] \\\n",
    "                .encode('utf-8'), topic_indices[i][1][j]))\n",
    "            \n",
    "\n",
    "    f.write(\"{0} topics distributed over {1} documents and {2} unique words\\n\"  \\\n",
    "        .format(num_topics, documents.count(), len(vocabulary)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server (127.0.0.1:49710)\n",
      "Traceback (most recent call last):\n",
      "  File \"/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/py4j/java_gateway.py\", line 827, in _get_connection\n",
      "    connection = self.deque.pop()\n",
      "IndexError: pop from an empty deque\n",
      "\n",
      "During handling of the above exception, another exception occurred:\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/py4j/java_gateway.py\", line 963, in start\n",
      "    self.socket.connect((self.address, self.port))\n",
      "ConnectionRefusedError: [Errno 61] Connection refused\n",
      "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/pyspark/context.py:413: RuntimeWarning: Unable to cleanly shutdown Spark JVM process. It is possible that the process has crashed, been killed or may also be in a zombie state.\n",
      "  RuntimeWarning\n"
     ]
    }
   ],
   "source": [
    "sc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "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.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}