part02/week2_160830/sparkR/Chap01_SecondarySort.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# CHAPTER 1  - Secondary Sort: Introduction\n",
    "\n",
    "- Secondary Sort는 key에 의한 정렬도 하고, key에 연결된 value에 대해서도 정렬하는 문제임.\n",
    "- MapReduce프레임워크 라는 병렬처리 알고리즘으로 해결하보자."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- MapReduce 기본 구조\n",
    "\n",
    "> map(key1, value1) → list(key2, value2)\n",
    "\n",
    "> reduce(key2, list(value2)) → list(key3, value3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Solutions to the Secondary Sort Problem\n",
    "\n",
    "\n",
    "첫번째 방법 \n",
    "\n",
    "- reduce단계에서 value값을 모두 메모리에 올려놓고 정렬함.\n",
    "- out-ofmemory error\n",
    "\n",
    "두번째 방법\n",
    "\n",
    "- MapReduce 프레임워크의 reducer의 정렬기능을 이용\n",
    "- key와 value을 하나의 key로 구성하는 composite key 생성이 필요\n",
    "- Hadoop의 MapReduce프레임워크만 가능하고, spark와 R에서 구현 불가\n",
    "\n",
    "\n",
    "![](sparkr_01_01.jpg)\n",
    "\n",
    "![](sparkr_01_02.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Spark Solution to Secondary Sort"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Time Series as Input\n",
    "\n",
    "```\n",
    "name time value\n",
    "x 2 9\n",
    "y 2 5\n",
    "x 1 3\n",
    "y 1 7\n",
    "y 3 1\n",
    "x 3 6\n",
    "z 1 4\n",
    "z 2 8\n",
    "z 3 7\n",
    "z 4 0\n",
    "p 2 6\n",
    "p 4 7\n",
    "p 1 9\n",
    "p 6 0\n",
    "p 7 3\n",
    "```\n",
    "\n",
    "### Expected Output\n",
    "\n",
    "```\n",
    "name t1 t2 t3 t4 t5 ...\n",
    "x => [3, 9, 6]\n",
    "y => [7, 5, 1]\n",
    "z => [4, 8, 7, 0]\n",
    "p => [9, 6, 7, 0, 3]\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3: Connect to the Spark master"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.context.SparkContext at 0x7fd68418ea50>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext() \n",
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 4: Use the JavaSparkContext to create a JavaRDD\n",
    "\n",
    "-  timeseries.csv  파일 내용\n",
    "```\n",
    "x,2,9\n",
    "y,2,5\n",
    "x,1,3\n",
    "y,1,7\n",
    "y,3,1\n",
    "x,3,6\n",
    "z,1,4\n",
    "z,2,8\n",
    "z,3,7\n",
    "z,4,0\n",
    "p,2,6\n",
    "p,4,7\n",
    "p,1,9\n",
    "p,6,0\n",
    "p,7,3\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lines = sc.textFile('timeseries.csv', 1);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'x,2,9',\n",
       " u'y,2,5',\n",
       " u'x,1,3',\n",
       " u'y,1,7',\n",
       " u'y,3,1',\n",
       " u'x,3,6',\n",
       " u'z,1,4',\n",
       " u'z,2,8',\n",
       " u'z,3,7',\n",
       " u'z,4,0']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines.take(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 5: Create key-value pairs from the JavaRDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def make_pairs( s ) : \n",
    "    tokens = s.split(\",\");\n",
    "    time = int(  tokens[1] )\n",
    "    value = int(  tokens[2] )\n",
    "    timevalue = ( time, value )\n",
    "    return ( tokens[0], timevalue )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "pairs = lines.map( make_pairs )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'x', (2, 9)),\n",
       " (u'y', (2, 5)),\n",
       " (u'x', (1, 3)),\n",
       " (u'y', (1, 7)),\n",
       " (u'y', (3, 1)),\n",
       " (u'x', (3, 6)),\n",
       " (u'z', (1, 4)),\n",
       " (u'z', (2, 8)),\n",
       " (u'z', (3, 7)),\n",
       " (u'z', (4, 0)),\n",
       " (u'p', (2, 6)),\n",
       " (u'p', (4, 7)),\n",
       " (u'p', (1, 9)),\n",
       " (u'p', (6, 0)),\n",
       " (u'p', (7, 3))]"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pairs.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 7: Group JavaPairRDD elements by the key (name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "groups = pairs.groupByKey();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'y', <pyspark.resultiterable.ResultIterable at 0x7fd65e5fed10>),\n",
       " (u'x', <pyspark.resultiterable.ResultIterable at 0x7fd65e5fed50>),\n",
       " (u'z', <pyspark.resultiterable.ResultIterable at 0x7fd65e5fe850>),\n",
       " (u'p', <pyspark.resultiterable.ResultIterable at 0x7fd65e5fe690>)]"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "groups.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "debug2 key=y\t value=(2, 5)(1, 7)(3, 1)\n",
      "debug2 key=x\t value=(2, 9)(1, 3)(3, 6)\n",
      "debug2 key=z\t value=(1, 4)(2, 8)(3, 7)(4, 0)\n",
      "debug2 key=p\t value=(2, 6)(4, 7)(1, 9)(6, 0)(7, 3)\n"
     ]
    }
   ],
   "source": [
    "debug2 = groups.collect()\n",
    "for t2 in  debug2 :\n",
    "    print \"debug2 key={}\\t value={}\".format( t2[0],  \"\".join([str(x) for x in t2[1]] )   )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 9: Sort the reducer’s values in memory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def secondary_sort( s ) :\n",
    "    \n",
    "    decorated = [(tup[0], tup) for tup in s ]\n",
    "    decorated.sort()\n",
    "    undecorated  = [tup for second, tup in decorated]\n",
    "    \n",
    "    return undecorated "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sorted = groups.mapValues( secondary_sort )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'y', [(1, 7), (2, 5), (3, 1)]),\n",
       " (u'x', [(1, 3), (2, 9), (3, 6)]),\n",
       " (u'z', [(1, 4), (2, 8), (3, 7), (4, 0)]),\n",
       " (u'p', [(1, 9), (2, 6), (4, 7), (6, 0), (7, 3)])]"
      ]
     },
     "execution_count": 80,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sorted.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 10: Key로 정렬함."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sorted2 = sorted.sortByKey(True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'p', [(1, 9), (2, 6), (4, 7), (6, 0), (7, 3)]),\n",
       " (u'x', [(1, 3), (2, 9), (3, 6)]),\n",
       " (u'y', [(1, 7), (2, 5), (3, 1)]),\n",
       " (u'z', [(1, 4), (2, 8), (3, 7), (4, 0)])]"
      ]
     },
     "execution_count": 86,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sorted2.collect()"
   ]
  },
  {
   "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.11"
  },
  "widgets": {
   "state": {},
   "version": "1.1.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}