part03/week01_160503/spark_04_key_value_pair.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 키/ 값 페어로 작업하기"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. 키/값 페어 RDD\n",
    "\n",
    "2. 파티셔닝\n",
    "\n",
    "## 배경\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 페어 RDD 생성\n",
    "\n",
    "다음 포맷은 키/값 데이터를 그대로 pair RDD로 만들어 리턴함.\n",
    "이를 위해 map() 함수가 쓰임.\n",
    "이 예제는 README를 읽어 각 라인의 첫 단어를 키로 데이터를 만드는 코드임.\n",
    "만들어진 데이터는 튜플로 기록 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "# lines=sc.textFile(\"README.md\")\n",
    "lines= sc.parallelize([\"holden likes coffee\", \"panda likes long strings and coffee\"])\n",
    "pairs = lines.map(lambda x: (x.split(\" \")[0],x))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 페어 RDD 트랜스포메이션\n",
    "\n",
    "#### 예( {(1,2), (3,4), (3,6)})\n",
    "|함수이름| 목적| 예| 결과|\n",
    "|---\n",
    "|reduceByKey(func)|동일 키에 대한 값을 합침|rdd.reduceByKey((x,y)=>x+y)| {(1,2),(3,10)}|\n",
    "|groupByKey()|동일 키에 대한 값을 그룹화|rdd.groupBykey()|{(1,[2]),(3,[4,6])|\n",
    "|combineByKey(createCombiner, mergeValue, mergeCombiners, partitioner)|다른 결과 타입으로 동일 키 값 합침|||\n",
    "|mapValues(func)|키 변경 없이 각 값에 func 적용|rdd.mapValues(x=>x+1)|{(1,3),(3,5),(3,7)}|\n",
    "|flatMapValues(func)|각 값에 반복자 함수 적용해서 리턴 값과 기존 키 페어를 만듦|rdd.flatMapValues(x=>(x to5))|{(1,2),(1,3),(1,4), (1,5),(3,4),(3,5)}| \n",
    "|keys()| 키값|rdd.key()|(1,3,3)|\n",
    "|values()| 밸류값|rdd.values()|(2,4,6)|\n",
    "|sortByKey()| 키로 정렬된 RDD|rdd.sortByKey()| {(1,2), (3,4), (3,6)}|\n",
    "\n",
    "\n",
    "#### 두 페어 RDD (rdd={(1,2), (3,4), (3,6)}, other={(3,9)}\n",
    "|함수이름| 목적| 예| 결과|\n",
    "|---\n",
    "|subtractByKey|다른쪽 RDD 키로 RDD데이터 삭제|rdd.subtractByKey(other)|{(1,2)}\n",
    "|join|inner join 수행|rdd.join(other)|{(3,(4,9)),(3,(6,9))}|\n",
    "|rightOuterJoin|오른쪽 RDD 키들을 대상으로 조인|rdd.rightOuterJoin(other)|{(3,(some(4),9)),(3,(some(6),9))}|\n",
    "|leftOuterJoin|왼쪽 RDD 키들을 대상으로 조인|rdd.leftOuterJoin(other)|{1,(2,None), (3,(some(4),9)),(3,(some(6),9))}|\n",
    "|cogroup|동일키의 양쪽 RDD를 그룹화|rdd.cogroup(other)| {(1,{[2],[]}, {(3,{[4],[6],[9]}| \n",
    "\n",
    "또한 기존 RDD 함수도 지원한다."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('holden', 'holden likes coffee')]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result = pairs.filter(lambda keyValue: len(keyValue[1])<20)\n",
    "result.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 집합연산\n",
    "\n",
    "reduceByKey() : reduce와 유사. 각 키와 키에 대해 합쳐진 값으로 새로운 RDD를 리턴\n",
    "\n",
    "foldByKey(): fold와 유사"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[96] at RDD at PythonRDD.scala:43"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd= sc.parallelize({(\"panda\",0), (\"pink\",3),(\"pirate\",3),(\"panda\",1),(\"pink\",4)})\n",
    "#rdd=sc.textFile(\"README.md\")\n",
    "rdd.mapValues(lambda x: (x,1).reduceByKey(lambda x, y: (x[0]+y[0],x[1]+y[1])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'', 67),\n",
       " (u'when', 1),\n",
       " (u'R,', 1),\n",
       " (u'including', 3),\n",
       " (u'computation', 1),\n",
       " (u'using:', 1),\n",
       " (u'guidance', 2),\n",
       " (u'Scala,', 1),\n",
       " (u'environment', 1),\n",
       " (u'only', 1),\n",
       " (u'rich', 1),\n",
       " (u'Apache', 1),\n",
       " (u'sc.parallelize(range(1000)).count()', 1),\n",
       " (u'Building', 1),\n",
       " (u'guide,', 1),\n",
       " (u'return', 2),\n",
       " (u'Please', 3),\n",
       " (u'Try', 1),\n",
       " (u'not', 1),\n",
       " (u'Spark', 13),\n",
       " (u'scala>', 1),\n",
       " (u'Note', 1),\n",
       " (u'cluster.', 1),\n",
       " (u'./bin/pyspark', 1),\n",
       " (u'params', 1),\n",
       " (u'through', 1),\n",
       " (u'GraphX', 1),\n",
       " (u'[run', 1),\n",
       " (u'abbreviated', 1),\n",
       " (u'[project', 2),\n",
       " (u'##', 8),\n",
       " (u'library', 1),\n",
       " (u'see', 1),\n",
       " (u'\"local\"', 1),\n",
       " (u'[Apache', 1),\n",
       " (u'will', 1),\n",
       " (u'#', 1),\n",
       " (u'processing,', 1),\n",
       " (u'for', 11),\n",
       " (u'[building', 1),\n",
       " (u'provides', 1),\n",
       " (u'print', 1),\n",
       " (u'supports', 2),\n",
       " (u'built,', 1),\n",
       " (u'[params]`.', 1),\n",
       " (u'available', 1),\n",
       " (u'run', 7),\n",
       " (u'tests](https://cwiki.apache.org/confluence/display/SPARK/Useful+Developer+Tools).',\n",
       "  1),\n",
       " (u'This', 2),\n",
       " (u'Hadoop,', 2),\n",
       " (u'Tests', 1),\n",
       " (u'example:', 1),\n",
       " (u'-DskipTests', 1),\n",
       " (u'Maven](http://maven.apache.org/).', 1),\n",
       " (u'programming', 1),\n",
       " (u'running', 1),\n",
       " (u'against', 1),\n",
       " (u'site,', 1),\n",
       " (u'comes', 1),\n",
       " (u'package.', 1),\n",
       " (u'and', 10),\n",
       " (u'package.)', 1),\n",
       " (u'prefer', 1),\n",
       " (u'documentation,', 1),\n",
       " (u'submit', 1),\n",
       " (u'tools', 1),\n",
       " (u'use', 3),\n",
       " (u'from', 1),\n",
       " (u'For', 2),\n",
       " (u'./bin/run-example', 2),\n",
       " (u'fast', 1),\n",
       " (u'systems.', 1),\n",
       " (u'<http://spark.apache.org/>', 1),\n",
       " (u'Hadoop-supported', 1),\n",
       " (u'way', 1),\n",
       " (u'README', 1),\n",
       " (u'MASTER', 1),\n",
       " (u'engine', 1),\n",
       " (u'building', 2),\n",
       " (u'usage', 1),\n",
       " (u'instance:', 1),\n",
       " (u'with', 3),\n",
       " (u'protocols', 1),\n",
       " (u'And', 1),\n",
       " (u'this', 1),\n",
       " (u'setup', 1),\n",
       " (u'shell:', 2),\n",
       " (u'project', 1),\n",
       " (u'following', 2),\n",
       " (u'distribution', 1),\n",
       " (u'detailed', 2),\n",
       " (u'have', 1),\n",
       " (u'stream', 1),\n",
       " (u'is', 6),\n",
       " (u'higher-level', 1),\n",
       " (u'tests', 2),\n",
       " (u'1000:', 2),\n",
       " (u'sample', 1),\n",
       " (u'[\"Specifying', 1),\n",
       " (u'Alternatively,', 1),\n",
       " (u'file', 1),\n",
       " (u'need', 1),\n",
       " (u'You', 3),\n",
       " (u'instructions.', 1),\n",
       " (u'different', 1),\n",
       " (u'programs,', 1),\n",
       " (u'storage', 1),\n",
       " (u'same', 1),\n",
       " (u'machine', 1),\n",
       " (u'Running', 1),\n",
       " (u'which', 2),\n",
       " (u'you', 4),\n",
       " (u'A', 1),\n",
       " (u'About', 1),\n",
       " (u'sc.parallelize(1', 1),\n",
       " (u'locally.', 1),\n",
       " (u'Hive', 2),\n",
       " (u'optimized', 1),\n",
       " (u'uses', 1),\n",
       " (u'Version\"](http://spark.apache.org/docs/latest/building-spark.html#specifying-the-hadoop-version)',\n",
       "  1),\n",
       " (u'variable', 1),\n",
       " (u'The', 1),\n",
       " (u'data', 1),\n",
       " (u'a', 8),\n",
       " (u'\"yarn\"', 1),\n",
       " (u'Thriftserver', 1),\n",
       " (u'processing.', 1),\n",
       " (u'./bin/spark-shell', 1),\n",
       " (u'Python', 2),\n",
       " (u'Spark](#building-spark).', 1),\n",
       " (u'clean', 1),\n",
       " (u'the', 21),\n",
       " (u'requires', 1),\n",
       " (u'talk', 1),\n",
       " (u'help', 1),\n",
       " (u'Hadoop', 3),\n",
       " (u'high-level', 1),\n",
       " (u'find', 1),\n",
       " (u'web', 1),\n",
       " (u'Shell', 2),\n",
       " (u'how', 2),\n",
       " (u'graph', 1),\n",
       " (u'run:', 1),\n",
       " (u'should', 2),\n",
       " (u'to', 14),\n",
       " (u'module,', 1),\n",
       " (u'given.', 1),\n",
       " (u'directory.', 1),\n",
       " (u'must', 1),\n",
       " (u'SparkPi', 2),\n",
       " (u'do', 2),\n",
       " (u'Programs', 1),\n",
       " (u'Many', 1),\n",
       " (u'YARN,', 1),\n",
       " (u'using', 2),\n",
       " (u'Example', 1),\n",
       " (u'Once', 1),\n",
       " (u'HDFS', 1),\n",
       " (u'Because', 1),\n",
       " (u'name', 1),\n",
       " (u'Testing', 1),\n",
       " (u'refer', 2),\n",
       " (u'Streaming', 1),\n",
       " (u'SQL', 2),\n",
       " (u'them,', 1),\n",
       " (u'analysis.', 1),\n",
       " (u'set', 2),\n",
       " (u'Scala', 2),\n",
       " (u'thread,', 1),\n",
       " (u'individual', 1),\n",
       " (u'examples', 2),\n",
       " (u'changed', 1),\n",
       " (u'runs.', 1),\n",
       " (u'Pi', 1),\n",
       " (u'More', 1),\n",
       " (u'Python,', 2),\n",
       " (u'Versions', 1),\n",
       " (u'its', 1),\n",
       " (u'version', 1),\n",
       " (u'wiki](https://cwiki.apache.org/confluence/display/SPARK).', 1),\n",
       " (u'`./bin/run-example', 1),\n",
       " (u'Configuration', 1),\n",
       " (u'command,', 2),\n",
       " (u'can', 6),\n",
       " (u'core', 1),\n",
       " (u'Guide](http://spark.apache.org/docs/latest/configuration.html)', 1),\n",
       " (u'MASTER=spark://host:7077', 1),\n",
       " (u'Documentation', 1),\n",
       " (u'downloaded', 1),\n",
       " (u'distributions.', 1),\n",
       " (u'Spark.', 1),\n",
       " (u'Spark\"](http://spark.apache.org/docs/latest/building-spark.html).', 1),\n",
       " (u'[\"Building', 1),\n",
       " (u'`examples`', 2),\n",
       " (u'on', 5),\n",
       " (u'package', 1),\n",
       " (u'of', 5),\n",
       " (u'APIs', 1),\n",
       " (u'pre-built', 1),\n",
       " (u'Big', 1),\n",
       " (u'or', 3),\n",
       " (u'learning,', 1),\n",
       " (u'locally', 2),\n",
       " (u'overview', 1),\n",
       " (u'one', 2),\n",
       " (u'(You', 1),\n",
       " (u'Online', 1),\n",
       " (u'versions', 1),\n",
       " (u'your', 1),\n",
       " (u'threads.', 1),\n",
       " (u'>>>', 1),\n",
       " (u'spark://', 1),\n",
       " (u'contains', 1),\n",
       " (u'system', 1),\n",
       " (u'start', 1),\n",
       " (u'build/mvn', 1),\n",
       " (u'basic', 1),\n",
       " (u'configure', 1),\n",
       " (u'that', 2),\n",
       " (u'N', 1),\n",
       " (u'\"local[N]\"', 1),\n",
       " (u'DataFrames,', 1),\n",
       " (u'particular', 2),\n",
       " (u'be', 2),\n",
       " (u'an', 3),\n",
       " (u'easiest', 1),\n",
       " (u'Interactive', 2),\n",
       " (u'cluster', 2),\n",
       " (u'page](http://spark.apache.org/documentation.html)', 1),\n",
       " (u'<class>', 1),\n",
       " (u'example', 3),\n",
       " (u'are', 1),\n",
       " (u'Data.', 1),\n",
       " (u'mesos://', 1),\n",
       " (u'computing', 1),\n",
       " (u'URL,', 1),\n",
       " (u'in', 5),\n",
       " (u'general', 2),\n",
       " (u'To', 2),\n",
       " (u'at', 2),\n",
       " (u'1000).count()', 1),\n",
       " (u'if', 4),\n",
       " (u'built', 1),\n",
       " (u'no', 1),\n",
       " (u'Java,', 1),\n",
       " (u'MLlib', 1),\n",
       " (u'also', 4),\n",
       " (u'other', 1),\n",
       " (u'build', 3),\n",
       " (u'online', 1),\n",
       " (u'several', 1),\n",
       " (u'[Configuration', 1),\n",
       " (u'class', 2),\n",
       " (u'programs', 2),\n",
       " (u'documentation', 3),\n",
       " (u'It', 2),\n",
       " (u'graphs', 1),\n",
       " (u'./dev/run-tests', 1),\n",
       " (u'first', 1),\n",
       " (u'latest', 1)]"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd=sc.textFile(\"README.md\")\n",
    "words=rdd.flatMap(lambda x: x.split(\" \"))\n",
    "result = words.map(lambda x: (x,1)).reduceByKey(lambda x,y: x+y)\n",
    "result.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "combineByKey(): 한 파티션 내의 데이터를 하나씩 읽어 새로운 데이터는 createCombiner()를 사용하여 해당 키의 어큐뮬레이터의 초기값을 만들고 기존 값이 있을 때는 mergeValue() 로 어큐뮬레이터의 현재 값에 새로운 값을 적용\n",
    "각 파티션에서 같은 키에 대해 여러 어큐뮬레이터가 작동할 수 있음. 이 때는 mergeCombiner()를 써서 합쳐짐.\n",
    "\n",
    "### 병렬화 수준 최적화\n",
    "모든 RDD는 고정된 개수의 파티션을 가지고 있으며 이것이 연산이 처리될 때 동시 작업의 수준을 결정하게 된다. 이 때 특정 개수의 파티션을 사용하도록 요구할 수 있다.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data = [(\"a\",3), (\"b\",4),(\"a\",1)]\n",
    "sc.parallelize(data.reduceByKey(lamda x, y: x+y,10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "repartition(): 파티셔닝 바꾸고 싶을 때(셔플링이 일어남)\n",
    "\n",
    "coalesce(): 파티션 개수 줄이는 경우에 데이터 이동이 일어나지 않음\n",
    "\n",
    "rdd.getNumPartitions(): 파티션 개수"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 데이터 그룹화\n",
    "groupByKey(): [K,Iterable[v]]\n",
    "cogroup(): 여러 RDD의 키를 공유해 그룹화 [K,(Iterable[v], Iterable[W])]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 조인\n",
    "내부 조인: 양쪽 RDD에 모두 존재하는 키만 결과가 됨.\n",
    "\n",
    "leftOuterJoin(): 원본 RDD의 키값과 합침.\n",
    "\n",
    "rightOuterJoin(): 키가 다른쪽 RDD에 존재해야 하며 원본 RDD에 옵션으로 표시\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 데이터 정렬\n",
    "\n",
    "sortByKey(): ascending이라는 인자를 받음"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('panda', 1), ('pink', 3), ('pirate', 3), ('panda', 0), ('pink', 4)]"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.sortByKey(ascending=True, numPartitions=None, keyfunc=lambda x:str(x))\n",
    "rdd.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 페어 RDD 액션\n",
    "\n",
    "countByKey(): 키 값 개수\n",
    "\n",
    "colectAsMap(): 결과를 맵 형태로\n",
    "\n",
    "lookup(key): 키에 대한 모든 값을 되돌려 줌"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 데이터 파티셔닝\n",
    "키의 모음들이 같은 노드에 함께 모이는 것을 지원\n",
    "\n",
    "### RDD의 파티셔너 정하기\n",
    "\n",
    "### 파티셔닝이 도움이 되는 연산들\n",
    "reduceByKey, combineByKey, lookup…\n",
    "• 각키에 대한 연산이 단일장비에서 이루어짐\n",
    "\n",
    "• cogorup, join …\n",
    "\n",
    "• 최소 하나 이상의 RDD가 네트워크를 통해 전송될 필요가 없게 해줌\n",
    "\n",
    "### 파티셔닝에 영향을 주는 연산들\n",
    "• 데이터를 파티션하는 연산에 의해 만들어진 RDD는 자동으로 설정됨\n",
    "\n",
    "   • ex) join: Hash 파티셔닝 됨\n",
    "\n",
    "• 지정된 파티셔닝을 쓰는 것이 보장되지 못하는 연산에서는 설정 X\n",
    "\n",
    "   • ex) map: 키의 변경 가능성이 존재\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 예제: 페이지 랭크\n",
    "\n",
    "각 페이지의 랭크를 1.0으로 초기화\n",
    "\n",
    "반복 주기마다 공헌치 {랭크(p) /이웃숫자(p)}를 이웃들에게 전송\n",
    "\n",
    "각 페이지 랭크를 갱신 0.15 + 0.85 * 받은 공헌치"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-52-43f70aa6fc27>, line 1)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;36m  File \u001b[1;32m\"<ipython-input-52-43f70aa6fc27>\"\u001b[1;36m, line \u001b[1;32m1\u001b[0m\n\u001b[1;33m    (/, 이웃, 리스트는, 스파크, 오브젝트에, 저장되어, 있다고, 가정)\u001b[0m\n\u001b[1;37m     ^\u001b[0m\n\u001b[1;31mSyntaxError\u001b[0m\u001b[1;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "// 이웃 리스트는 스파크 오브젝트에 저장되어 있다고 가정\n",
    "val links = sc.objectFile[(String, Seq[String])](\"links\")\n",
    "              .partitionBy(new HashPartitioner(100))\n",
    "              .persist()\n",
    "\n",
    "        // 각 페이지의 기본 랭크를 1.0으로 초기화\n",
    "var ranks = links.mapValues(v => 1.0)\n",
    "\n",
    "// 알고리즘 10회 반복\n",
    "for (i <- 0 until 10) {\n",
    "    val contributions = links.join(ranks).flatMap {\n",
    "        case (pageId, ( links , rank)) =>\n",
    "            links .map(dest => (dest, rank / links.size))\n",
    "}\n",
    "ranks = contributions.reduceByKey((x, y) => x + y).mapValues(v => 0.15 + 0.85*v)\n",
    "}\n",
    "// 결과 저장\n",
    "ranks.saveAsTextFile(\"ranks\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 사용자 지정 파티셔너\n",
    "같은 도메인의 안의 링크를 동일한 파티션으로 지정\n",
    "\n",
    "org.apache.spark.Partitioner 상속\n",
    "\n",
    "• numPartitions: 파티션 수\n",
    "\n",
    "• getPartiton: 키에 대한 파티션 ID 반환\n",
    "\n",
    "• equlas: 두 RDD가 같은 방식으로 파티션 되었는지 검사"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "class DomainNamePartitioner(numParts: Int) extends Partitioner {\n",
    "    override def numPartitions: Int = numParts\n",
    "    override def getPartition(key: Any): Int = {\n",
    "        val domain = new Java.net.URL(key.toString).getHost()\n",
    "        val code = (domain.hashCode % numPartitions)\n",
    "        if (code < 0) {\n",
    "            code + numPartitions // 음수인 경우 0 이상인 값으로\n",
    "        } else {\n",
    "          code\n",
    "        }\n",
    "    }\n",
    "    // 자바의 equals 메소드, 스파크가 직접 만든 파티셔너 객체를 비교하는데 사용\n",
    "    override def equals(other: Any): Boolean = other match {\n",
    "        case dnp: DomainNamePartitioner =>\n",
    "            dnp.numPartitions == numPartitions\n",
    "        case _ =>\n",
    "        false\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[113] at mapPartitions at PythonRDD.scala:374"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import urlparse\n",
    "\n",
    "def hash_domain(url):\n",
    "    return hash(urlparse.urlparse(url).netloc)\n",
    "\n",
    "rdd.partitionBy(20,hash_domain)"
   ]
  },
  {
   "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"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}