Final.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import sys"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark_path = \"D:/spark\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "os.environ['SPARK_HOME'] = spark_path"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "os.environ['HADOOP_HOME'] = spark_path"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "os.environ['PYSPARK_PYTHON'] = sys.executable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sys.path.append(spark_path + \"/bin\")\n",
    "sys.path.append(spark_path + \"/python\")\n",
    "sys.path.append(spark_path + \"/python/pyspark/\")\n",
    "sys.path.append(spark_path + \"/python/lib\")\n",
    "sys.path.append(spark_path + \"/python/lib/pyspark.zip\")\n",
    "sys.path.append(spark_path + \"/python/lib/py4j-0.10.4-src.zip\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark import SparkConf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.conf.SparkConf at 0x5cc6f60>"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf = SparkConf()\n",
    "conf.set(\"spark.executor.memory\", \"2g\")\n",
    "conf.set(\"spark.cores.max\", \"4\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc = SparkContext(\"local\", conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<pyspark.context.SparkContext object at 0x0000000006956710>\n"
     ]
    }
   ],
   "source": [
    "print sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "airlinesPath=\"C:\\\\Users\\\\ifirf\\Documents\\\\BigData\\\\airlines.csv\"\n",
    "airportsPath=\"C:\\\\Users\\ifirf\\\\Documents\\\\BigData\\\\airport.csv\"\n",
    "flightsPath=\"C:\\\\Users\\\\ifirf\\\\Documents\\\\BigData\\\\flights.csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "flights = sc.textFile(flightsPath)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "476881"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flights.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'2014-04-01,19805,1,JFK,LAX,0854,-6.00,1217,2.00,355.00,2475.00',\n",
       " u'2014-04-01,19805,2,LAX,JFK,0944,14.00,1736,-29.00,269.00,2475.00',\n",
       " u'2014-04-01,19805,3,JFK,LAX,1224,-6.00,1614,39.00,371.00,2475.00',\n",
       " u'2014-04-01,19805,4,LAX,JFK,1240,25.00,2028,-27.00,264.00,2475.00',\n",
       " u'2014-04-01,19805,5,DFW,HNL,1300,-5.00,1650,15.00,510.00,3784.00',\n",
       " u'2014-04-01,19805,6,OGG,DFW,1901,126.00,0640,95.00,385.00,3711.00',\n",
       " u'2014-04-01,19805,7,DFW,OGG,1410,125.00,1743,138.00,497.00,3711.00',\n",
       " u'2014-04-01,19805,8,HNL,DFW,1659,4.00,0458,-22.00,398.00,3784.00',\n",
       " u'2014-04-01,19805,9,JFK,LAX,0648,-7.00,1029,19.00,365.00,2475.00',\n",
       " u'2014-04-01,19805,10,LAX,JFK,2156,21.00,0556,1.00,265.00,2475.00']"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flights.take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "flightsParsed=flights.map(lambda x:x.split(\",\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "class Flight(tuple):\n",
      "    'Flight(date, airline, flightnum, origin, dest, dep, dep_delay, arv, arv_delay, airtime, distance)'\n",
      "\n",
      "    __slots__ = ()\n",
      "\n",
      "    _fields = ('date', 'airline', 'flightnum', 'origin', 'dest', 'dep', 'dep_delay', 'arv', 'arv_delay', 'airtime', 'distance')\n",
      "\n",
      "    def __new__(_cls, date, airline, flightnum, origin, dest, dep, dep_delay, arv, arv_delay, airtime, distance):\n",
      "        'Create new instance of Flight(date, airline, flightnum, origin, dest, dep, dep_delay, arv, arv_delay, airtime, distance)'\n",
      "        return _tuple.__new__(_cls, (date, airline, flightnum, origin, dest, dep, dep_delay, arv, arv_delay, airtime, distance))\n",
      "\n",
      "    @classmethod\n",
      "    def _make(cls, iterable, new=tuple.__new__, len=len):\n",
      "        'Make a new Flight object from a sequence or iterable'\n",
      "        result = new(cls, iterable)\n",
      "        if len(result) != 11:\n",
      "            raise TypeError('Expected 11 arguments, got %d' % len(result))\n",
      "        return result\n",
      "\n",
      "    def __repr__(self):\n",
      "        'Return a nicely formatted representation string'\n",
      "        return 'Flight(date=%r, airline=%r, flightnum=%r, origin=%r, dest=%r, dep=%r, dep_delay=%r, arv=%r, arv_delay=%r, airtime=%r, distance=%r)' % self\n",
      "\n",
      "    def _asdict(self):\n",
      "        'Return a new OrderedDict which maps field names to their values'\n",
      "        return OrderedDict(zip(self._fields, self))\n",
      "\n",
      "    def _replace(_self, **kwds):\n",
      "        'Return a new Flight object replacing specified fields with new values'\n",
      "        result = _self._make(map(kwds.pop, ('date', 'airline', 'flightnum', 'origin', 'dest', 'dep', 'dep_delay', 'arv', 'arv_delay', 'airtime', 'distance'), _self))\n",
      "        if kwds:\n",
      "            raise ValueError('Got unexpected field names: %r' % kwds.keys())\n",
      "        return result\n",
      "\n",
      "    def __getnewargs__(self):\n",
      "        'Return self as a plain tuple.  Used by copy and pickle.'\n",
      "        return tuple(self)\n",
      "\n",
      "    __dict__ = _property(_asdict)\n",
      "\n",
      "    def __getstate__(self):\n",
      "        'Exclude the OrderedDict from pickling'\n",
      "        pass\n",
      "\n",
      "    date = _property(_itemgetter(0), doc='Alias for field number 0')\n",
      "\n",
      "    airline = _property(_itemgetter(1), doc='Alias for field number 1')\n",
      "\n",
      "    flightnum = _property(_itemgetter(2), doc='Alias for field number 2')\n",
      "\n",
      "    origin = _property(_itemgetter(3), doc='Alias for field number 3')\n",
      "\n",
      "    dest = _property(_itemgetter(4), doc='Alias for field number 4')\n",
      "\n",
      "    dep = _property(_itemgetter(5), doc='Alias for field number 5')\n",
      "\n",
      "    dep_delay = _property(_itemgetter(6), doc='Alias for field number 6')\n",
      "\n",
      "    arv = _property(_itemgetter(7), doc='Alias for field number 7')\n",
      "\n",
      "    arv_delay = _property(_itemgetter(8), doc='Alias for field number 8')\n",
      "\n",
      "    airtime = _property(_itemgetter(9), doc='Alias for field number 9')\n",
      "\n",
      "    distance = _property(_itemgetter(10), doc='Alias for field number 10')\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from datetime import datetime\n",
    "from collections import namedtuple\n",
    "\n",
    "fields = ('date', 'airline', 'flightnum', 'origin', 'dest', 'dep', \n",
    "          'dep_delay', 'arv', 'arv_delay', 'airtime', 'distance')\n",
    "Flight = namedtuple('Flight', fields, verbose=True)\n",
    "\n",
    "DATE_FMT =\"%Y-%m-%d\"\n",
    "TIME_FMT = \"%H%M\"\n",
    "\n",
    "def parse(row):\n",
    "    row[0] = datetime.strptime(row[0], DATE_FMT).date()\n",
    "    row[5] = datetime.strptime(row[5], TIME_FMT).time()\n",
    "    row[6] = float(row[6])\n",
    "    row[7] = datetime.strptime(row[7], TIME_FMT).time()\n",
    "    row[8] = float(row[8])\n",
    "    row[9] = float(row[9])\n",
    "    row[10] = float(row[10])\n",
    "    return Flight(*row[:11])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "flightsParsed=flights.map(lambda x:x.split(\",\")).map(parse)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Flight(date=datetime.date(2014, 4, 1), airline=u'19805', flightnum=u'1', origin=u'JFK', dest=u'LAX', dep=datetime.time(8, 54), dep_delay=-6.0, arv=datetime.time(12, 17), arv_delay=2.0, airtime=355.0, distance=2475.0)"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flightsParsed.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "totalDistance=flightsParsed.map(lambda x:x.distance).reduce(lambda x,y:x+y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "avgDistance=totalDistance/flightsParsed.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "794.858501387\n"
     ]
    }
   ],
   "source": [
    "print avgDistance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.3753871510922012"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flightsParsed.filter(lambda x:x.dep_delay>0).count()/float(flightsParsed.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[12] at RDD at PythonRDD.scala:48"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flightsParsed.persist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sumCount=flightsParsed.map(lambda x:x.dep_delay).aggregate((0,0),\n",
    "                                                          (lambda acc,value: (acc[0]+value, acc[1]+1)),\n",
    "                                                          (lambda acc1,acc2:(acc1[0]+acc2[0],acc1[1]+acc2[1])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The average delay is 8.31387704689\n"
     ]
    }
   ],
   "source": [
    "print \"The average delay is \" + str (sumCount[0]/float(sumCount[1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {0: 452963,\n",
       "             1: 16016,\n",
       "             2: 4893,\n",
       "             3: 1729,\n",
       "             4: 701,\n",
       "             5: 249,\n",
       "             6: 113,\n",
       "             7: 66,\n",
       "             8: 43,\n",
       "             9: 26,\n",
       "             10: 15,\n",
       "             11: 12,\n",
       "             12: 9,\n",
       "             13: 15,\n",
       "             14: 13,\n",
       "             15: 4,\n",
       "             17: 2,\n",
       "             20: 4,\n",
       "             21: 3,\n",
       "             24: 3,\n",
       "             25: 1,\n",
       "             28: 1})"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flightsParsed.map(lambda x:int(x.dep_delay/60)).countByValue()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}