performance-tuning.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark as ps\n",
    "import json\n",
    "from pyspark.sql import HiveContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc = ps.SparkContext(master='spark://127.0.0.1:7077', conf=ps.SparkConf().setAppName(\"Performance Tuning\"))\n",
    "hive_context = HiveContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "# need to get local path since we are reading local files\n",
    "cwd = os.getcwd()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "file_rdd = sc.textFile('file://' + cwd + '/data/toy_data.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# default OSX block size is 4kb\n",
    "file_rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\"Jane\": \"2\"}\r\n",
      "{\"Jane\": \"1\"}\r\n",
      "{\"Pete\": \"20\"}\r\n",
      "{\"Tyler\": \"3\"}\r\n",
      "{\"Duncan\": \"4\"}\r\n",
      "{\"Yuki\": \"5\"}\r\n",
      "{\"Duncan\": \"6\"}\r\n",
      "{\"Duncan\": \"4\"}\r\n",
      "{\"Duncan\": \"5\"}\r\n"
     ]
    }
   ],
   "source": [
    "cat data/toy_data.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0,\n",
       " [u'{\"Jane\": \"2\"}',\n",
       "  u'{\"Jane\": \"1\"}',\n",
       "  u'{\"Pete\": \"20\"}',\n",
       "  u'{\"Tyler\": \"3\"}',\n",
       "  u'{\"Duncan\": \"4\"}'],\n",
       " 1,\n",
       " [u'{\"Yuki\": \"5\"}',\n",
       "  u'{\"Duncan\": \"6\"}',\n",
       "  u'{\"Duncan\": \"4\"}',\n",
       "  u'{\"Duncan\": \"5\"}']]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "file_rdd.mapPartitionsWithIndex(lambda i, iterator: (i, list(iterator))).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "file_rdd = sc.textFile('file:///Users/jonathandinu/spark-ds-applications/data/toy_data.txt', 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0,\n",
       " [u'{\"Jane\": \"2\"}'],\n",
       " 1,\n",
       " [u'{\"Jane\": \"1\"}'],\n",
       " 2,\n",
       " [u'{\"Pete\": \"20\"}'],\n",
       " 3,\n",
       " [u'{\"Tyler\": \"3\"}'],\n",
       " 4,\n",
       " [u'{\"Duncan\": \"4\"}'],\n",
       " 5,\n",
       " [u'{\"Yuki\": \"5\"}'],\n",
       " 6,\n",
       " [u'{\"Duncan\": \"6\"}'],\n",
       " 7,\n",
       " [u'{\"Duncan\": \"4\"}'],\n",
       " 8,\n",
       " [],\n",
       " 9,\n",
       " [u'{\"Duncan\": \"5\"}'],\n",
       " 10,\n",
       " []]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# some partitions are empty, can't split a single line\n",
    "file_rdd.mapPartitionsWithIndex(lambda i, iterator: (i, list(iterator))).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "tup_rdd = file_rdd.map(lambda line: json.loads(line)) \\\n",
    "                  .map(lambda d: (d.keys()[0], int(d.values()[0])))\n",
    "\n",
    "out = tup_rdd.groupByKey().mapValues(lambda tup: max(tup.data)) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "toy_data_tuples PythonRDD[10] at RDD at PythonRDD.scala:43"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tup_rdd.setName('toy_data_tuples')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'Jane', 2),\n",
       " (u'Jane', 1),\n",
       " (u'Pete', 20),\n",
       " (u'Tyler', 3),\n",
       " (u'Duncan', 4),\n",
       " (u'Yuki', 5),\n",
       " (u'Duncan', 6),\n",
       " (u'Duncan', 4),\n",
       " (u'Duncan', 5)]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tup_rdd.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(11) PythonRDD[11] at RDD at PythonRDD.scala:43 []\n",
      " |   MapPartitionsRDD[9] at mapPartitions at PythonRDD.scala:346 []\n",
      " |   ShuffledRDD[8] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      " +-(11) PairwiseRDD[7] at groupByKey at <ipython-input-9-5dd19bb9f251>:3 []\n",
      "    |   PythonRDD[6] at groupByKey at <ipython-input-9-5dd19bb9f251>:3 []\n",
      "    |   MapPartitionsRDD[4] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      "    |   file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt HadoopRDD[3] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print out.toDebugString()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Readychef"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "meals_rdd = sc.textFile('file:///Users/jonathandinu/spark-ds-applications/data/readychef/meals.txt')\n",
    "events_rdd = sc.textFile('file:///Users/jonathandinu/spark-ds-applications/data/readychef/events.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "header_meals = meals_rdd.first()\n",
    "header_events = events_rdd.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "meals_no_header = meals_rdd.filter(lambda row: row != header_meals)\n",
    "events_no_header = events_rdd.filter(lambda row: row != header_events)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "meals_json = meals_no_header.map(lambda row: row.split(';')) \\\n",
    "                            .map(lambda row_list: dict(zip(header_meals.split(';'), row_list)))\n",
    "    \n",
    "events_json = events_no_header.map(lambda row: row.split(';')) \\\n",
    "                              .map(lambda row_list: dict(zip(header_events.split(';'), row_list)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[124] at RDD at PythonRDD.scala:43"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "meals_json.cache()\n",
    "events_json.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def type_conversion(d, columns):\n",
    "    for c in columns:\n",
    "        d[c] = int(d[c])\n",
    "        \n",
    "    return d"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "meals_typed = meals_json.map(lambda j: json.dumps(type_conversion(j, ['meal_id', 'price'])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "events_typed = events_json.map(lambda j: json.dumps(type_conversion(j, ['meal_id', 'userid'])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'italian', 22575),\n",
       " (u'french', 16179),\n",
       " (u'mexican', 8792),\n",
       " (u'japanese', 6921),\n",
       " (u'chinese', 6267),\n",
       " (u'vietnamese', 3535)]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# no pre-filter\n",
    "sc.setJobGroup('PySpark -- nofilter', \"PySpark nofilter join performance\")\n",
    "\n",
    "meals_json.keyBy(lambda x: x['meal_id']) \\\n",
    "            .join(events_json.keyBy(lambda x: x['meal_id'])) \\\n",
    "            .filter(lambda tup: tup[1][1]['event'] == 'bought') \\\n",
    "            .groupBy(lambda tup: tup[1][0]['type']) \\\n",
    "            .mapValues(lambda val: len(val)) \\\n",
    "            .sortBy(lambda g: g[1], ascending=False) \\\n",
    "            .collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'italian', 22575),\n",
       " (u'french', 16179),\n",
       " (u'mexican', 8792),\n",
       " (u'japanese', 6921),\n",
       " (u'chinese', 6267),\n",
       " (u'vietnamese', 3535)]"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pre-filter before join\n",
    "sc.setJobGroup('PySpark -- prefilter', \"PySpark cached RDD join performance\")\n",
    "\n",
    "meals_json.keyBy(lambda x: x['meal_id']) \\\n",
    "            .join(events_json.filter(lambda x: x['event'] == 'bought') \\\n",
    "                             .keyBy(lambda x: x['meal_id']) \\\n",
    "                 ) \\\n",
    "            .filter(lambda tup: tup[1][1]['event'] == 'bought') \\\n",
    "            .groupBy(lambda tup: tup[1][0]['type']) \\\n",
    "            .mapValues(lambda val: len(val)) \\\n",
    "            .sortBy(lambda g: g[1], ascending=False) \\\n",
    "            .collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'italian', 22575),\n",
       " (u'french', 16179),\n",
       " (u'mexican', 8792),\n",
       " (u'japanese', 6921),\n",
       " (u'chinese', 6267),\n",
       " (u'vietnamese', 3535)]"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# reduceByKey\n",
    "sc.setJobGroup('PySpark -- prefilter', \"PySpark cached RDD join with reduce by key performance\")\n",
    "\n",
    "meals_json.keyBy(lambda x: x['meal_id']) \\\n",
    "            .join(events_json.filter(lambda x: x['event'] == 'bought') \\\n",
    "                             .keyBy(lambda x: x['meal_id']) \\\n",
    "                 ) \\\n",
    "            .filter(lambda tup: tup[1][1]['event'] == 'bought') \\\n",
    "            .map(lambda tup: (tup[1][0]['type'], 1)) \\\n",
    "            .reduceByKey(lambda a, b: a + b) \\\n",
    "            .sortBy(lambda g: g[1], ascending=False) \\\n",
    "            .collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'164',\n",
       "  ({u'dt': u'2013-01-31',\n",
       "    u'meal_id': u'164',\n",
       "    u'price': u'15',\n",
       "    u'type': u'italian'},\n",
       "   {u'dt': u'2013-01-27',\n",
       "    u'event': u'bought',\n",
       "    u'meal_id': u'164',\n",
       "    u'userid': u'76'})),\n",
       " (u'164',\n",
       "  ({u'dt': u'2013-01-31',\n",
       "    u'meal_id': u'164',\n",
       "    u'price': u'15',\n",
       "    u'type': u'italian'},\n",
       "   {u'dt': u'2013-01-29',\n",
       "    u'event': u'bought',\n",
       "    u'meal_id': u'164',\n",
       "    u'userid': u'31'})),\n",
       " (u'164',\n",
       "  ({u'dt': u'2013-01-31',\n",
       "    u'meal_id': u'164',\n",
       "    u'price': u'15',\n",
       "    u'type': u'italian'},\n",
       "   {u'dt': u'2013-01-29',\n",
       "    u'event': u'bought',\n",
       "    u'meal_id': u'164',\n",
       "    u'userid': u'88'})),\n",
       " (u'164',\n",
       "  ({u'dt': u'2013-01-31',\n",
       "    u'meal_id': u'164',\n",
       "    u'price': u'15',\n",
       "    u'type': u'italian'},\n",
       "   {u'dt': u'2013-01-29',\n",
       "    u'event': u'bought',\n",
       "    u'meal_id': u'164',\n",
       "    u'userid': u'203'})),\n",
       " (u'164',\n",
       "  ({u'dt': u'2013-01-31',\n",
       "    u'meal_id': u'164',\n",
       "    u'price': u'15',\n",
       "    u'type': u'italian'},\n",
       "   {u'dt': u'2013-01-30',\n",
       "    u'event': u'bought',\n",
       "    u'meal_id': u'164',\n",
       "    u'userid': u'179'}))]"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "joined_rdd.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(4) PythonRDD[90] at collect at <ipython-input-37-1cdbd71232ce>:1 []\n",
      " |  MapPartitionsRDD[89] at mapPartitions at PythonRDD.scala:346 []\n",
      " |  ShuffledRDD[88] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      " +-(4) PairwiseRDD[87] at sortBy at <ipython-input-36-8c75a2437747>:1 []\n",
      "    |  PythonRDD[86] at sortBy at <ipython-input-36-8c75a2437747>:1 []\n",
      "    |  MapPartitionsRDD[83] at mapPartitions at PythonRDD.scala:346 []\n",
      "    |  ShuffledRDD[82] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      "    +-(4) PairwiseRDD[81] at groupBy at <ipython-input-35-cfb2ef5806c3>:1 []\n",
      "       |  PythonRDD[80] at groupBy at <ipython-input-35-cfb2ef5806c3>:1 []\n",
      "       |  MapPartitionsRDD[79] at mapPartitions at PythonRDD.scala:346 []\n",
      "       |  ShuffledRDD[78] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      "       +-(4) PairwiseRDD[77] at join at <ipython-input-34-5071b7146a50>:4 []\n",
      "          |  PythonRDD[76] at join at <ipython-input-34-5071b7146a50>:4 []\n",
      "          |  UnionRDD[75] at union at NativeMethodAccessorImpl.java:-2 []\n",
      "          |  PythonRDD[73] at RDD at PythonRDD.scala:43 []\n",
      "          |  MapPartitionsRDD[14] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      "          |  file:///Users/jonathandinu/spark-ds-applications/data/readychef/meals.txt HadoopRDD[13] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      "          |  PythonRDD[74] at RDD at PythonRDD.scala:43 []\n",
      "          |  MapPartitionsRDD[16] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      "          |  file:///Users/jonathandinu/spark-ds-applications/data/readychef/events.txt HadoopRDD[15] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print ordered_list.toDebugString()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc.setJobGroup('Spark SQL', \"spark sql performance\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "meals_dataframe = hive_context.jsonRDD(meals_typed)\n",
    "events_dataframe = hive_context.jsonRDD(events_typed)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "meals_dataframe.registerTempTable('meals')\n",
    "events_dataframe.registerTempTable('events')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(type=u'italian', cnt=22575),\n",
       " Row(type=u'french', cnt=16179),\n",
       " Row(type=u'mexican', cnt=8792),\n",
       " Row(type=u'japanese', cnt=6921),\n",
       " Row(type=u'chinese', cnt=6267),\n",
       " Row(type=u'vietnamese', cnt=3535)]"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# which cuisine sells the most\n",
    "\n",
    "hive_context.sql(\"\"\"\n",
    "    SELECT type, COUNT(type) as cnt FROM\n",
    "        meals \n",
    "    INNER JOIN \n",
    "        events on meals.meal_id = events.meal_id\n",
    "    WHERE\n",
    "        event = 'bought'\n",
    "    GROUP BY\n",
    "        type\n",
    "    ORDER BY cnt DESC\n",
    "\"\"\").collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Airlines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "sc.setJobGroup('Airline Data', \"Airline Dataset\")\n",
    "\n",
    "link = 's3n://[AWS_ACCESS_KEY]:[AWS_SECRET]@mortar-example-data/airline-data'\n",
    "airline = sc.textFile(link)\n",
    "\n",
    "airline_no_quote = airline.map(lambda line: line.replace('\\'', '').replace('\\\"', '').strip(','))\n",
    "\n",
    "header_line = airline_no_quote.first()\n",
    "header_list = header_line.split(',')\n",
    "\n",
    "airline_no_header = airline_no_quote.filter(lambda row: row != header_line)\n",
    "\n",
    "def make_row(row):\n",
    "    row_list = row.split(',')\n",
    "    \n",
    "    d = dict(zip(header_list, row_list))\n",
    "    \n",
    "    return d\n",
    "\n",
    "airline_rows = airline_no_header.map(make_row)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(u'12129', -6.7547169811320753), (u'15991', -6.0978441127694856), (u'12888', -5.9056603773584904), (u'14113', -5.3462002412545235), (u'10779', -5.1457627118644069), (u'13127', -5.0891265597147948), (u'14633', -4.9087677725118484), (u'10739', -4.666666666666667), (u'15897', -4.6107142857142858), (u'11274', -4.6034482758620694)]\n"
     ]
    }
   ],
   "source": [
    "sc.setJobGroup('Airline Data', \"no filter\")\n",
    "\n",
    "destination_rdd = airline_rows.map(lambda row: (row['DEST_AIRPORT_ID'], row))\n",
    "origin_rdd = airline_rows.map(lambda row: (row['ORIGIN_AIRPORT_ID'], row))\n",
    "\n",
    "mean_delays_dest = destination_rdd.groupByKey() \\\n",
    "                                  .mapValues(lambda delays: \\\n",
    "                                             np.mean(map(lambda row: \\\n",
    "                                                             float(row['ARR_DELAY']) if row['ARR_DELAY'] else 0, \\\n",
    "                                                         delays.data)))\n",
    "    \n",
    "mean_delays_origin = origin_rdd.groupByKey()\n",
    "                               .mapValues(lambda delays: \\\n",
    "                                          np.mean(map(lambda row: \\\n",
    "                                                          float(row['DEP_DELAY']) if row['DEP_DELAY'] else 0, \\\n",
    "                                                      delays.data)))\n",
    "\n",
    "print mean_delays_origin.sortBy(lambda t: t[1], ascending=True).take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(u'12129', -6.7547169811320753), (u'15991', -6.0978441127694856), (u'12888', -5.9056603773584904), (u'14113', -5.3462002412545235), (u'10779', -5.1457627118644069), (u'13127', -5.0891265597147948), (u'14633', -4.9087677725118484), (u'10739', -4.666666666666667), (u'15897', -4.6107142857142858), (u'11274', -4.6034482758620694)]\n"
     ]
    }
   ],
   "source": [
    "sc.setJobGroup('Airline Data', \"filtered first\")\n",
    "\n",
    "destination_rdd = airline_rows.map(lambda row: (row['DEST_AIRPORT_ID'], \\\n",
    "                                               float(row['ARR_DELAY'] if row['ARR_DELAY'] else 0)))\n",
    "\n",
    "origin_rdd = airline_rows.map(lambda row: (row['ORIGIN_AIRPORT_ID'], \\\n",
    "                                          float(row['DEP_DELAY']) if row['DEP_DELAY'] else 0))\n",
    "\n",
    "mean_delays_dest = destination_rdd.groupByKey().mapValues(lambda delays: np.mean(delays.data))\n",
    "mean_delays_origin = origin_rdd.groupByKey().mapValues(lambda delays: np.mean(delays.data))\n",
    "\n",
    "print mean_delays_origin.sortBy(lambda t: t[1], ascending=True).take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "airline_rows.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(11) PythonRDD[122] at RDD at PythonRDD.scala:43 []\n",
      " |   MapPartitionsRDD[114] at mapPartitions at PythonRDD.scala:346 []\n",
      " |   ShuffledRDD[113] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      " +-(11) PairwiseRDD[112] at groupByKey at <ipython-input-28-824c2b202e95>:6 []\n",
      "    |   PythonRDD[111] at groupByKey at <ipython-input-28-824c2b202e95>:6 []\n",
      "    |   MapPartitionsRDD[90] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      "    |   s3n://AKIAJRN6IWDBI5XT3M6Q:3iOWT533UlP1gzzJeIYiixCZLDcYeUA9g0wAWGET@mortar-example-data/airline-data HadoopRDD[89] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print mean_delays_origin.toDebugString()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(u'12402', (-6601.0, 6001)), (u'11898', (-5898.0, 1510)), (u'14113', (-4432.0, 829)), (u'14633', (-4143.0, 844)), (u'11648', (-3880.0, 1832)), (u'15991', (-3677.0, 603)), (u'12280', (-3646.0, 2664)), (u'10779', (-3036.0, 590)), (u'13127', (-2855.0, 561)), (u'11695', (-2795.0, 1892))]\n"
     ]
    }
   ],
   "source": [
    "sc.setJobGroup('Airline Data -- filtered', \"reduceByKey + filtered\")\n",
    "\n",
    "destination_rdd = airline_rows.map(lambda row: (row['DEST_AIRPORT_ID'], \\\n",
    "                                               float(row['ARR_DELAY'] if row['ARR_DELAY'] else 0)))\n",
    "\n",
    "origin_rdd = airline_rows.map(lambda row: (row['ORIGIN_AIRPORT_ID'], \\\n",
    "                                          float(row['DEP_DELAY']) if row['DEP_DELAY'] else 0))\n",
    "\n",
    "mean_delays_dest = destination_rdd.mapValues(lambda row: (row, 1)) \\\n",
    "                                  .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "    \n",
    "mean_delays_origin = origin_rdd.mapValues(lambda row: (row, 1)) \\\n",
    "                               .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "\n",
    "print mean_delays_origin.sortBy(lambda t: t[1], ascending=True).take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[218] at RDD at PythonRDD.scala:43"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "airline_rows.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(u'12402', (-6601.0, 6001)), (u'11898', (-5898.0, 1510)), (u'14113', (-4432.0, 829)), (u'14633', (-4143.0, 844)), (u'11648', (-3880.0, 1832)), (u'15991', (-3677.0, 603)), (u'12280', (-3646.0, 2664)), (u'10779', (-3036.0, 590)), (u'13127', (-2855.0, 561)), (u'11695', (-2795.0, 1892))]\n"
     ]
    }
   ],
   "source": [
    "sc.setJobGroup('Airline Data -- filtered', \"cache first run\")\n",
    "\n",
    "destination_rdd = airline_rows.map(lambda row: (row['DEST_AIRPORT_ID'], \\\n",
    "                                               float(row['ARR_DELAY'] if row['ARR_DELAY'] else 0)))\n",
    "\n",
    "origin_rdd = airline_rows.map(lambda row: (row['ORIGIN_AIRPORT_ID'], \\\n",
    "                                          float(row['DEP_DELAY']) if row['DEP_DELAY'] else 0))\n",
    "\n",
    "mean_delays_dest = destination_rdd.mapValues(lambda row: (row, 1)) \\\n",
    "                                  .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "    \n",
    "mean_delays_origin = origin_rdd.mapValues(lambda row: (row, 1)) \\\n",
    "                               .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "\n",
    "print mean_delays_origin.sortBy(lambda t: t[1], ascending=True).take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(u'12402', (-6601.0, 6001)), (u'11898', (-5898.0, 1510)), (u'14113', (-4432.0, 829)), (u'14633', (-4143.0, 844)), (u'11648', (-3880.0, 1832)), (u'15991', (-3677.0, 603)), (u'12280', (-3646.0, 2664)), (u'10779', (-3036.0, 590)), (u'13127', (-2855.0, 561)), (u'11695', (-2795.0, 1892))]\n"
     ]
    }
   ],
   "source": [
    "sc.setJobGroup('Airline Data -- filtered', \"cache second run\")\n",
    "\n",
    "destination_rdd = airline_rows.map(lambda row: (row['DEST_AIRPORT_ID'], \\\n",
    "                                               float(row['ARR_DELAY'] if row['ARR_DELAY'] else 0)))\n",
    "\n",
    "origin_rdd = airline_rows.map(lambda row: (row['ORIGIN_AIRPORT_ID'], \\\n",
    "                                          float(row['DEP_DELAY']) if row['DEP_DELAY'] else 0))\n",
    "\n",
    "mean_delays_dest = destination_rdd.mapValues(lambda row: (row, 1)) \\\n",
    "                                  .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "    \n",
    "mean_delays_origin = origin_rdd.mapValues(lambda row: (row, 1)) \\\n",
    "                               .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "\n",
    "print mean_delays_origin.sortBy(lambda t: t[1], ascending=True).take(10)"
   ]
  }
 ],
 "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.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}