Step 0 - Convert GNIP to Per User GeoJSON.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json, iso8601, pprint, os, codecs, pandas as pd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load, Clean, Sort, Convert GNIP Tweets to Per User GeoJSON\n",
    "\n",
    "This notebook:\n",
    "\n",
    "- Loads raw gnip data\n",
    "- Writes full GNIP GEOJSONL files per user\n",
    "\n",
    "[Spark Status](http://epic-analytics.cs.colorado.edu:4040/)\n",
    "\n",
    "## EVENT CONFIGURATION"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(Timestamp('2017-08-25 00:00:00+0000', tz='UTC'),\n",
       " Timestamp('2017-08-27 00:00:00+0000', tz='UTC'))"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "input_directory  = \"/data/chime/Hurricanes_HIM/geo/harvey/GNIP/\"\n",
    "output_directory = \"/data/chime/geo2/HARVEY/\"\n",
    "\n",
    "MINIMUM_TWEET_THRESHOLD = 5\n",
    "STORM_PERIOD_START = pd.Timestamp(\"2017-08-25T00:00:00Z\")\n",
    "STORM_PERIOD_END   = pd.Timestamp(\"2017-08-27T00:00:00Z\")\n",
    "(STORM_PERIOD_START, STORM_PERIOD_END)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Load the files\n",
    "raw_strings = sc.textFile(input_directory)\n",
    "#Filter out the duds\n",
    "strings = raw_strings.filter(lambda x: x!=\"\")\n",
    "#JSONs\n",
    "jsons  = strings.map(json.loads)\n",
    "tweets = jsons.filter(lambda x: 'info' not in x.keys())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 2: Group tweets by user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[16] at RDD at PythonRDD.scala:48"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tweets_gb_user = tweets.groupBy(lambda t: t['actor']['id'])\n",
    "tweets_gb_user.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Force Spark to Execute..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "#tweets_gb_user.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check on the status of this operation, should see a tuple of: (user_id, iterable)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "#pprint.pprint(tweets_gb_user.take(1)[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3: Make GeoJSON"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "def writeGeoJSON(uTuple):\n",
    "    u, iterable = uTuple\n",
    "    tweets = list(iterable)\n",
    "    \n",
    "    #Break if there are not enough tweets\n",
    "    if len(tweets) < MINIMUM_TWEET_THRESHOLD:\n",
    "        return False\n",
    "    \n",
    "    handle = tweets[0]['actor']['preferredUsername']\n",
    "    tweets.sort(key=lambda t: t['postedTime'])\n",
    "    \n",
    "    features = []\n",
    "    in_storm_range_with_geo = 0;\n",
    "    \n",
    "    for t in tweets:\n",
    "        try:\n",
    "            geo = t.get('geo',None)\n",
    "            if geo:\n",
    "                time = pd.Timestamp(t['postedTime'])\n",
    "                if (time > STORM_PERIOD_START and time < STORM_PERIOD_END):\n",
    "                    in_storm_range_with_geo += 1;\n",
    "                geo = {'type':\"Point\",'coordinates':list(reversed(geo['coordinates']))}\n",
    "            feat = {\n",
    "                'type':'Feature',\n",
    "                'geometry': geo,\n",
    "                'properties':{\n",
    "                    'user':handle, #Use the original handle incase a name changed\n",
    "                    'text':t['body'],\n",
    "                    'date':t['postedTime'],\n",
    "                    'tweetID': t['id'].split(\":\")[2]\n",
    "                }\n",
    "            }\n",
    "            features.append(feat)\n",
    "        except:\n",
    "            raise\n",
    "    \n",
    "    #Minimum tweet count?\n",
    "    if in_storm_range_with_geo >= MINIMUM_TWEET_THRESHOLD:\n",
    "        json.dump({'type':'FeatureCollection', 'features': features},\n",
    "                  codecs.open(output_directory + handle+\".geojson\",'w'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3: Write out the geojson files (per user)\n",
    "Very basic simplified tweets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "if not os.path.exists(output_directory):\n",
    "    os.mkdir(output_directory)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "tweets_gb_user.foreach( writeGeoJSON )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "pySpark (Spark 2.2.1)",
   "language": "python3",
   "name": "pyspark"
  },
  "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}