Twitter Mongo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "![Twitter](https://upload.wikimedia.org/wikipedia/commons/thumb/5/51/Twitter_logo.svg/469px-Twitter_logo.svg.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import twitter\n",
    "import nltk\n",
    "import time\n",
    "\n",
    "from pymongo import MongoClient\n",
    "\n",
    "__version__ = '1.0'\n",
    "__all__ = []\n",
    "__author__ = 'Axel Oehmichen - ao1011@imperial.ac.uk'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def prepare_tweet(tweet_json):\n",
    "    tweet_dic = dict(tweet_json)\n",
    "    user_id = tweet_dic[\"user\"][\"id\"]\n",
    "    tweet_dic[\"user\"] =str(user_id)\n",
    "    return tweet_dic\n",
    "\n",
    "def insert_timeline_into_mongo(twitter_user_id, api, MONGO_URL):\n",
    "\n",
    "    mongo_client = MongoClient(MONGO_URL )\n",
    "    users_collection = mongo_client.twitter.twitterUsers\n",
    "    tweets_collection = mongo_client.twitter.tweets\n",
    "\n",
    "    max_id = None  # since_id parameter to the greatest ID of all the Tweets your application has already processed.\n",
    "    count = 200  # We retrieve 200 tweets at a time\n",
    "    current_count = 0\n",
    "    max_count = 1500\n",
    "    timeline_json = []\n",
    "    # Spark output isn't idempotent so, although the insert occurs in a transaction,\n",
    "    # it's possible for it to succeed in both tasks before one can be cancelled.\n",
    "    if users_collection.find({\"user.id\": twitter_user_id}).count() == 0:\n",
    "        users_collection.insert_one({\"id\": twitter_user_id})\n",
    "        while current_count <= max_count:\n",
    "            # We retrieve the first chunk of tweets\n",
    "            timeline_chunk = api.GetUserTimeline(twitter_user_id, max_id=max_id, count=count)\n",
    "            if len(timeline_chunk) == 1 :\n",
    "                current_count = max_count + 1\n",
    "            else:\n",
    "                max_id = timeline_chunk[-1].id\n",
    "                timeline_json.extend(timeline_chunk)               \n",
    "                # We insert the tweets into the collection\n",
    "                tweets_collection.insert_many([ prepare_tweet(timeline_chunk[i]._json)  for i in range(len(timeline_chunk))])\n",
    "                current_count += len(timeline_chunk)\n",
    "                time.sleep(1)\n",
    "        \n",
    "        # We insert our user to the user collection\n",
    "        users_collection.update_one({'id': twitter_user_id}, {\"$set\":  dict(timeline_json[0].user._json)}, upsert=False)\n",
    "    # we close our mongo connection\n",
    "    mongo_client.close()\n",
    "    return \"OK\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def process_user(twitter_user_id, api, MONGO_URL):\n",
    "\n",
    "    tweets_inserted_status = insert_timeline_into_mongo(twitter_user_id, api, MONGO_URL)\n",
    "    \n",
    "    return (twitter_user_id , tweets_inserted_status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Main Program\n",
    "\n",
    "We will now set all the paramters required to access twitter and the MongoDb database."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Twitter key and secret for OAuth\n",
    "consumer_key = \"XXX\"\n",
    "consumer_secret = \"YYY\"\n",
    "\n",
    "access_token = \"AAA\"\n",
    "access_token_secret = \"BBB\"\n",
    "\n",
    "api = twitter.Api(consumer_key=consumer_key,\n",
    "                  consumer_secret=consumer_secret,\n",
    "                  access_token_key=access_token,\n",
    "                  access_token_secret=access_token_secret)\n",
    "\n",
    "# The users chosen are\n",
    "user_ids = [\"25073877\", \"813286\", \"1339835893\", \"52544275\", \"409486555\", \"759251\", \"3235334092\"]\n",
    "\n",
    "# Address of the mongo cluster\n",
    "MONGO_URL = \"mongodb://\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We retieve the timelines for the specified users and print out \"OK\" when the task is completed by the worker."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "users_ids_rdd = sc.parallelize(user_ids)\n",
    "insertion = users_ids_rdd.map(lambda user_id : process_user(user_id, api, MONGO_URL))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "insertion.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Natural Language Processing\n",
    "\n",
    "### We do a small language processing on the tweets and we insert them back into a new collection.\n",
    "If you are interested in discovering further the nltk library : http://www.nltk.org/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def process_tweets_for_user(twitter_user_id, MONGO_URL):\n",
    "    mongo_client = MongoClient(MONGO_URL )\n",
    "    tweets_collection = mongo_client.twitter.tweets\n",
    "    tweets_processed = mongo_client.twitter.processedTweets\n",
    "\n",
    "    for tweet in tweets_collection.find({\"user\": twitter_user_id}):\n",
    "        text = tweet[\"text\"]\n",
    "        tokens = nltk.word_tokenize(text)\n",
    "        tagged = nltk.pos_tag(tokens)\n",
    "        doc = {\"text\": text,\n",
    "               \"tokens\": tokens,\n",
    "               \"tagged\": tagged\n",
    "               }\n",
    "        tweets_processed.insert_one(doc)\n",
    "    return \"Processed\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "process_status = users_ids_rdd.map(lambda user_id : process_tweets_for_user(user_id, MONGO_URL))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "process_status.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Exercises\n",
    "\n",
    "**The reference documentation for pymongo is available at that address:** https://api.mongodb.com/python/current/ \n",
    "\n",
    "Queries:\n",
    "* Count the number of tweets and users\n",
    "* Print out the name of all the users inserted\n",
    "* Find the most retweeted tweet\n",
    "* Find the shortest tweet\n",
    "* Count all the words used the tweets and find the top 5 most used\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "mongo_client = MongoClient(MONGO_URL)\n",
    "\n",
    "# Count the number of tweets \n",
    "mongo_client.twitter.tweets.count()\n",
    "\n",
    "# Count the number of users\n",
    "mongo_client.twitter.twitterUsers.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Print out the name of all the users inserted \n",
    "cursor = mongo_client.twitter.twitterUsers.find({},{\"name\": 1})\n",
    "\n",
    "for document in cursor:\n",
    "    print document[\"name\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pymongo import DESCENDING \n",
    "# Find the most retweeted tweet\n",
    "most_retweeted = mongo_client.twitter.tweets.find().sort(\"retweet_count\", DESCENDING).limit(1)\n",
    "\n",
    "for t in most_retweeted:\n",
    "    print t[\"retweet_count\"]\n",
    "    print t"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Find the shortest tweet\n",
    "def text_length(tweet_id, MONGO_URL):\n",
    "    mongo_client = MongoClient(MONGO_URL)\n",
    "    tweet = mongo_client.twitter.tweets.find_one({\"_id\": tweet_id[\"_id\"]})\n",
    "    return (tweet[\"text\"], len(tweet[\"text\"]))\n",
    "\n",
    "def compare_length(tweet1, tweet2):\n",
    "    if(tweet1[1]<tweet2[1]):\n",
    "        return  tweet1\n",
    "    else:\n",
    "        return tweet2\n",
    "\n",
    "tweets_ids = sc.parallelize(list(mongo_client.twitter.tweets.find({},{\"_id\": 1})))\n",
    "\n",
    "shortest_tweet = tweets_ids.map(lambda tweet_id : text_length(tweet_id, MONGO_URL)).reduce(lambda t1,t2: compare_length(t1,t2))\n",
    "\n",
    "print \"The shortest tweet is \" + str(shortest_tweet[0]) + \"\\nAnd the length is \" +  str(shortest_tweet[1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Count all the words used the tweets and find the top 5 most used\n",
    "from bson.code import Code\n",
    "mapper = Code(\"\"\"\n",
    "               function () {\n",
    "                 this.tokens.forEach(function(z) {\n",
    "                   emit(z, 1);\n",
    "                 });\n",
    "               }\n",
    "               \"\"\")\n",
    "\n",
    "reducer = Code(\"\"\"\n",
    "                function (key, values) {\n",
    "                  var total = 0;\n",
    "                  for (var i = 0; i < values.length; i++) {\n",
    "                    total += values[i];\n",
    "                  }\n",
    "                  return total;\n",
    "                }\n",
    "                \"\"\")\n",
    "\n",
    "result =  mongo_client.twitter.processedTweets.map_reduce(mapper, reducer, \"myresults\")\n",
    "\n",
    "for doc in result.find().sort(\"value\", DESCENDING).limit(5):\n",
    "    print doc"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "python",
   "name": "pyspark"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}