spark-internals.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark as ps\n",
    "sc = ps.SparkContext()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "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": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'{\"Jane\": \"2\"}', u'{\"Jane\": \"1\"}']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "file_rdd = sc.textFile('file://' + cwd + '/data/toy_data.txt')\n",
    "file_rdd.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# default OSX block size is 4kb\n",
    "file_rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-rwxr-xr-x@ 1 jonathandinu  staff   136B Aug  5 14:12 \u001b[31mdata/toy_data.txt\u001b[m\u001b[m\r\n"
     ]
    }
   ],
   "source": [
    "!ls -larth data/toy_data.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "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": 33,
   "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": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "file_rdd.mapPartitionsWithIndex(lambda i, iterator: (i, list(iterator))).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "file_rdd = sc.textFile('file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt', 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "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": 39,
     "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": 40,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "file_rdd = sc.textFile('file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt', 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "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",
       "  u'{\"Yuki\": \"5\"}',\n",
       "  u'{\"Duncan\": \"6\"}',\n",
       "  u'{\"Duncan\": \"4\"}',\n",
       "  u'{\"Duncan\": \"5\"}']]"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "file_rdd.mapPartitionsWithIndex(lambda i, iterator: (i, list(iterator))).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "collapsed": false
   },
   "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": 50,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) PythonRDD[78] at RDD at PythonRDD.scala:43 []\n",
      " |  MapPartitionsRDD[71] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      " |  file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt HadoopRDD[70] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print tup_rdd.toDebugString()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) PythonRDD[77] at RDD at PythonRDD.scala:43 []\n",
      " |  MapPartitionsRDD[76] at mapPartitions at PythonRDD.scala:346 []\n",
      " |  ShuffledRDD[75] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      " +-(2) PairwiseRDD[74] at groupByKey at <ipython-input-48-5dd19bb9f251>:3 []\n",
      "    |  PythonRDD[73] at groupByKey at <ipython-input-48-5dd19bb9f251>:3 []\n",
      "    |  MapPartitionsRDD[71] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      "    |  file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt HadoopRDD[70] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print out.toDebugString()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) MapPartitionsRDD[71] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      " |  file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt HadoopRDD[70] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print file_rdd.toDebugString()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) PythonRDD[85] at RDD at PythonRDD.scala:43 []\n",
      " |  MapPartitionsRDD[71] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      " |  file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt HadoopRDD[70] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "# doesn't matter if 1 or 100 map transformations\n",
    "print file_rdd.map(lambda line: json.loads(line)).toDebugString()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) PythonRDD[95] at RDD at PythonRDD.scala:43 []\n",
      " |  MapPartitionsRDD[71] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      " |  file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt HadoopRDD[70] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print tup_rdd.toDebugString()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) PythonRDD[90] at RDD at PythonRDD.scala:43 []\n",
      " |  MapPartitionsRDD[89] at mapPartitions at PythonRDD.scala:346 []\n",
      " |  ShuffledRDD[88] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      " +-(2) PairwiseRDD[87] at groupByKey at <ipython-input-60-62a481cc790d>:1 []\n",
      "    |  PythonRDD[86] at groupByKey at <ipython-input-60-62a481cc790d>:1 []\n",
      "    |  MapPartitionsRDD[71] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      "    |  file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt HadoopRDD[70] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print tup_rdd.groupByKey().toDebugString()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) PythonRDD[96] at RDD at PythonRDD.scala:43 []\n",
      " |  MapPartitionsRDD[94] at mapPartitions at PythonRDD.scala:346 []\n",
      " |  ShuffledRDD[93] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      " +-(2) PairwiseRDD[92] at groupByKey at <ipython-input-61-5dd19bb9f251>:3 []\n",
      "    |  PythonRDD[91] at groupByKey at <ipython-input-61-5dd19bb9f251>:3 []\n",
      "    |  MapPartitionsRDD[71] at textFile at NativeMethodAccessorImpl.java:-2 []\n",
      "    |  file:///Users/jonathandinu/Repositories/pearson/data-applications-sprk/clean_notebooks/data/toy_data.txt HadoopRDD[70] at textFile at NativeMethodAccessorImpl.java:-2 []\n"
     ]
    }
   ],
   "source": [
    "print out.toDebugString()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Scala"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```scala\n",
    "file_rdd.mapPartitionsWithIndex((ind, part) => part.map(x => (ind, x))).collect()\n",
    "\n",
    "file_rdd.map(line => line.split(\" \"))\n",
    "        .map(split => (split(0), split(1).toInt))\n",
    "        .groupByKey()\n",
    "        .mapValues(iter => iter.reduce(_ + _)).collect()\n",
    "\n",
    "file_rdd.map(line => line.split(\" \")).map(split => (split(0), split(1).toInt))\n",
    "                                     .groupByKey()\n",
    "                                     .mapPartitionsWithIndex((ind, part) => part.map(x => (ind, x)))\n",
    "                                     .collect()\n",
    "                                     \n",
    "// res68: Array[(Int, (String, Iterable[Int]))] = Array((0,(anna,CompactBuffer(1))), (0,(jesse,CompactBuffer(3))), (1,(jon,CompactBuffer(2, 1))), (1,(mary,CompactBuffer(3, 5))))\n",
    "```"
   ]
  }
 ],
 "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
}