Zone A Calculations (Spark).ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "ename": "ImportError",
     "evalue": "No module named shapely",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m\u001b[0m",
      "\u001b[0;31mImportError\u001b[0mTraceback (most recent call last)",
      "\u001b[0;32m<ipython-input-4-62b03cae174f>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mosgeo\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mogr\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0;32mimport\u001b[0m \u001b[0mshapely\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mfiona\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mjson\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;31mImportError\u001b[0m: No module named shapely"
     ]
    }
   ],
   "source": [
    "from osgeo import ogr\n",
    "import shapely, fiona, json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "raw_strings = sc.textFile(\"result_data/all_tweets_one_file.jsonl\") # Would Using HDFS be smarter here?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'{\"geometry\": {\"type\": \"Point\", \"coordinates\": [-74.5342206, 39.3954102]}, \"type\": \"Feature\", \"properties\": {\"u_utc\": null, \"text\": \"- I do my own thing , you made me this way .\", \"source\": {\"link\": \"http://twitter.com/download/android\", \"displayName\": \"Twitter for Android\"}, \"verb\": \"post\", \"user\": \"Dougieee_Nastyy\", \"time\": \"2012-10-25T11:25:26.000Z\", \"gnip\": {\"matching_rules\": [{\"tag\": \"[-74.53321912339669,39.36436374931048]\", \"value\": \"bounding_box:[-74.57813488760267 39.329638110038545 -74.48830335919072 39.39908938858241]\"}], \"language\": {\"value\": \"en\"}}}}']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "raw_strings.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[5] at RDD at PythonRDD.scala:53"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tweets = raw_strings.map(json.loads)\n",
    "tweets.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1 = tweets.take(1)[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{u'geometry': {u'type': u'Point', u'coordinates': [-74.5342206, 39.3954102]}, u'type': u'Feature', u'properties': {u'u_utc': None, u'text': u'- I do my own thing , you made me this way .', u'source': {u'displayName': u'Twitter for Android', u'link': u'http://twitter.com/download/android'}, u'verb': u'post', u'user': u'Dougieee_Nastyy', u'time': u'2012-10-25T11:25:26.000Z', u'gnip': {u'matching_rules': [{u'tag': u'[-74.53321912339669,39.36436374931048]', u'value': u'bounding_box:[-74.57813488760267 39.329638110038545 -74.48830335919072 39.39908938858241]'}], u'language': {u'value': u'en'}}}}\n"
     ]
    }
   ],
   "source": [
    "print(t1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load Zone A as a Geometry"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from shapely.geometry import mapping, shape\n",
    "c = fiona.open('ZoneA_Geometry/ZoneA/OGRGeoJSON.shp','r')\n",
    "pol = c.next()\n",
    "geom = shape(pol['geometry']).buffer(0)\n",
    "geom.is_valid"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "zoneABroadcast = sc.broadcast(geom)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'tweets_with_geo' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m\u001b[0m",
      "\u001b[0;31mNameError\u001b[0mTraceback (most recent call last)",
      "\u001b[0;32m<ipython-input-1-bdc864c8c256>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mtest_tweets\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mtweets_with_geo\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtake\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;36m10\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m: name 'tweets_with_geo' is not defined"
     ]
    }
   ],
   "source": [
    "test_tweets = tweets_with_geo.take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "POINT (-73.949 40.73389)\n",
      "True\n",
      "POINT (-73.99753114562988 40.73093368341445)\n",
      "False\n"
     ]
    }
   ],
   "source": [
    "#Testing:\n",
    "in_bounds = shape({'type': \"Point\", 'coordinates': [-73.94900, 40.73389]})\n",
    "print in_bounds\n",
    "print( geom.contains(in_bounds))\n",
    "# print(\"In Bounds (Should be True): \" + str( point_in_bounds.Within(zoneABroadcast.value)))\n",
    "\n",
    "out_of_bounds = shape({\"type\": \"Point\",\"coordinates\": [-73.99753114562988,40.73093368341445]})\n",
    "print out_of_bounds\n",
    "print geom.contains(out_of_bounds)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "POINT (-74.5342206 39.3954102)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<shapely.geometry.point.Point at 0x7f58040c18d0>"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def get_point(tweet):\n",
    "    tweet['geo']['coordinates'].reverse()\n",
    "    return shape(tweet['geo'])\n",
    "\n",
    "#Test it:\n",
    "for t in test_tweets:\n",
    "p = get_point(t1)\n",
    "print(p)\n",
    "p"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Use Spark to find ZoneA tweets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "inZoneA = tweets.filter(lambda t: zoneABroadcast.value.contains(get_point(t)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "110111"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Kick off the Action, save the tweets\n",
    "inZoneA.cache()\n",
    "inZoneA.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "inZoneAStrings = inZoneA.map(json.dumps)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "inZoneAStrings.saveAsTextFile('inZoneA.geojsonl.hdfs')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Users: 22042\n"
     ]
    }
   ],
   "source": [
    "grouped_by_user = inZoneA.groupBy(lambda t: t['properties']['user'])\n",
    "count_by_user   = grouped_by_user.map(lambda tu: (tu[0],len(tu[1]) ))\n",
    "sorted_by_quant = count_by_user.sortBy(lambda tu: -tu[1])\n",
    "sorted_by_quant.cache()\n",
    "print(\"Users: \" + str(sorted_by_quant.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write out all the tweets to GeoJSON\n",
    "\n",
    "# Now build a secondary GeoJSON server to only serve specific dates?\n",
    "\n",
    "# for tu in sorted_by_quant.collect():\n",
    "#     u_tweets = tweets.filter(lambda t: t['properties']['user'] == tu[0])\n",
    "#     features = u_tweets.collect()\n",
    "#     with open('zone_a_twitterers/'+tu[0]+'.geojson','w') as outGeo:\n",
    "#         json.dump({\"type\":\"FeatureCollection\",\"features\":features},outGeo)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Temporal Filtering of Zone A Tweets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json, datetime, iso8601, pytz"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def between_two_dates(tweet, start, end):\n",
    "        time = iso8601.parse_date(tweet['properties']['time'])\n",
    "#         del tweet['properties']['gnip']\n",
    "        return time > start and time < end"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "LANDFALL = datetime.datetime(2012,10,29,20,0,0,tzinfo=pytz.utc) #Landfall as UTC\n",
    "AFTER    = datetime.datetime(2012,11,1,0,0,0,tzinfo=pytz.utc)   #After as UTC\n",
    "\n",
    "inZoneA_landfall = inZoneA.filter( lambda tweet: between_two_dates(tweet, LANDFALL, AFTER) )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2466"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "inZoneA_landfall.cache()\n",
    "inZoneA_landfall.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Users: 916\n"
     ]
    }
   ],
   "source": [
    "zoneALandfall_grouped_by_user = inZoneA_landfall.groupBy(lambda t: t['properties']['user'])\n",
    "zoneALandfall_count_by_user   = zoneALandfall_grouped_by_user.map(lambda tu: (tu[0],len(tu[1]) ))\n",
    "zoneALandfall_sorted_by_quant = zoneALandfall_count_by_user.sortBy(lambda tu: -tu[1])\n",
    "zoneALandfall_sorted_by_quant.cache()\n",
    "print(\"Users: \" + str(zoneALandfall_sorted_by_quant.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'Zav718', 106),\n",
       " (u'hermaflease', 90),\n",
       " (u'jennifferkkwilt', 90),\n",
       " (u'charlenexcfye', 86),\n",
       " (u'innycru', 56),\n",
       " (u'mr0drig', 55),\n",
       " (u'laynaBrady', 33),\n",
       " (u'honeyberk', 31),\n",
       " (u'lucida_console', 25),\n",
       " (u'carminenyc', 23),\n",
       " (u'Omg_itsClyde', 20),\n",
       " (u'JenniferCochra1', 17),\n",
       " (u'JoeyBoots', 17),\n",
       " (u'laurakazam', 17),\n",
       " (u'salvadorsorian5', 16),\n",
       " (u'Sarahphara', 15),\n",
       " (u'GeorgieeeNinjaa', 14),\n",
       " (u'anneeoanneo', 14),\n",
       " (u'frankgiraldi', 14),\n",
       " (u'BobSpivak1', 13),\n",
       " (u'mcflipper', 13),\n",
       " (u'fefaromano', 13),\n",
       " (u'epjohanson', 13),\n",
       " (u'eelain212', 12),\n",
       " (u'rahav', 11),\n",
       " (u'JUSTmeALISEO', 10),\n",
       " (u'johnjohnta', 10),\n",
       " (u'AdamGabbatt', 10),\n",
       " (u'kirilnyc', 9),\n",
       " (u'cpthady', 9),\n",
       " (u'j_livoti', 9),\n",
       " (u'mmmpork', 9),\n",
       " (u'M3lissaSays', 9),\n",
       " (u'bgallis', 9),\n",
       " (u'AnthonyQuintano', 9),\n",
       " (u'JayDensonNYC', 9),\n",
       " (u'DaReal_JrJones', 9),\n",
       " (u'smoshysydney', 9),\n",
       " (u'julienneschaer', 8),\n",
       " (u'JonMcL', 8),\n",
       " (u'JamieAlexandraa', 8),\n",
       " (u'marky_mark34', 8),\n",
       " (u'taftcard', 8),\n",
       " (u'LucaTrippitelli', 8),\n",
       " (u'MrsPuertoRico', 8),\n",
       " (u'whistlesandbeer', 8),\n",
       " (u'elisesp', 8),\n",
       " (u'savorymango', 8),\n",
       " (u'therealgowanus', 7),\n",
       " (u'noahsussman', 7),\n",
       " (u'molly_mcgregor', 7),\n",
       " (u'thekenmendoza', 7),\n",
       " (u'LaLaHearts', 7),\n",
       " (u'ArnellMilton', 7),\n",
       " (u'gkor29', 7),\n",
       " (u'kat_pugacheva', 6),\n",
       " (u'mashikMe', 6),\n",
       " (u'JOtton', 6),\n",
       " (u'ChaosKnight69', 6),\n",
       " (u'nicklaw01', 6),\n",
       " (u'conrad', 6),\n",
       " (u'angeljimenez', 6),\n",
       " (u'dem_CALLmi_tutu', 6),\n",
       " (u'eunicek', 6),\n",
       " (u'Sparks_Photo', 6),\n",
       " (u'arojass', 6),\n",
       " (u'steketee', 6),\n",
       " (u'artuhovski', 5),\n",
       " (u'LightoutsRock', 5),\n",
       " (u'cooper_smith', 5),\n",
       " (u'JCelona_', 5),\n",
       " (u'jmmantel', 5),\n",
       " (u'hands9me', 5),\n",
       " (u'iamamayzing', 5),\n",
       " (u'RAjah1', 5),\n",
       " (u'Maderised', 5),\n",
       " (u'areda1120', 5),\n",
       " (u'daveevansabc7', 5),\n",
       " (u'DCamz5', 5),\n",
       " (u'xomegsss', 5),\n",
       " (u'itsAdam', 5),\n",
       " (u'JonCowley', 5),\n",
       " (u'jcm_mejia91', 5),\n",
       " (u'rav617', 5),\n",
       " (u'JamesBruni', 4),\n",
       " (u'ProfPizza', 4),\n",
       " (u'Todd_SimmonsNYC', 4),\n",
       " (u'tvartboy', 4),\n",
       " (u'CrowleyMonster', 4),\n",
       " (u'IrishDaveNYSpur', 4),\n",
       " (u'GaryAlonyNYC', 4),\n",
       " (u'christiannseka', 4),\n",
       " (u'knight_kat', 4),\n",
       " (u'ericbogs', 4),\n",
       " (u'alcazareo', 4),\n",
       " (u'g0ttalov3him', 4),\n",
       " (u'cfohlin', 4),\n",
       " (u'urifintzy', 4),\n",
       " (u'BAHjournalist', 4),\n",
       " (u'poppyphi', 4)]"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "zoneALandfall_sorted_by_quant.take(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "pySpark (Spark 1.5.2)",
   "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.14"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}