RDD.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark Context\n",
    "- spark app 과 cluster 연결 관리하는 객체"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "lastException = null\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Name: org.apache.spark.SparkException\n",
       "Message: Only one SparkContext may be running in this JVM (see SPARK-2243). To ignore this error, set spark.driver.allowMultipleContexts = true. The currently running SparkContext was created at:\n",
       "org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:860)\n",
       "org.apache.toree.kernel.api.Kernel.sparkSession(Kernel.scala:425)\n",
       "org.apache.toree.kernel.api.Kernel.sparkContext(Kernel.scala:429)\n",
       "$line8.$read$$iw$$iw$$iw$$iw.sc(<console>:17)\n",
       "$line13.$read$$iw$$iw$$iw$$iw$$iw$$iw.<init>(<console>:27)\n",
       "$line13.$read$$iw$$iw$$iw$$iw$$iw.<init>(<console>:40)\n",
       "$line13.$read$$iw$$iw$$iw$$iw.<init>(<console>:42)\n",
       "$line13.$read$$iw$$iw$$iw.<init>(<console>:44)\n",
       "$line13.$read$$iw$$iw.<init>(<console>:46)\n",
       "$line13.$read$$iw.<init>(<console>:48)\n",
       "$line13.$read.<init>(<console>:50)\n",
       "$line13.$read$.<init>(<console>:54)\n",
       "$line13.$read$.<clinit>(<console>)\n",
       "$line13.$eval$.$print$lzycompute(<console>:7)\n",
       "$line13.$eval$.$print(<console>:6)\n",
       "$line13.$eval.$print(<console>)\n",
       "sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n",
       "sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n",
       "sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
       "java.lang.reflect.Method.invoke(Method.java:498)\n",
       "StackTrace:   at org.apache.spark.SparkContext$$anonfun$assertNoOtherContextIsRunning$2.apply(SparkContext.scala:2274)\n",
       "  at scala.Option.foreach(Option.scala:257)\n",
       "  at org.apache.spark.SparkContext$.assertNoOtherContextIsRunning(SparkContext.scala:2274)\n",
       "  at org.apache.spark.SparkContext$.markPartiallyConstructed(SparkContext.scala:2353)\n",
       "  at org.apache.spark.SparkContext.<init>(SparkContext.scala:85)"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.SparkConf\n",
    "import org.apache.spark.SparkContext\n",
    "import org.apache.spark.SparkContext._\n",
    "\n",
    "val conf = new SparkConf().setMaster(\"local\").setAppName(\"RddSample\")\n",
    "val sc1 = new SparkContext(conf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.3 RDD 생성\n",
    "\n",
    "- 드라이버 프로그램의 collection object 이용하는 방법"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[0] at parallelize at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[0] at parallelize at <console>:33"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc1.parallelize(List(\"a\", \"b\", \"c\", \"d\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 파일이나 데이터 데이스 같은 외부 데이터를 읽어서 새로운 RDD 생성하는 방법\n",
    "\n",
    "    - textFile의 파일 각줄은 한 개의 RDD 구성요소가 됨"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd2 = RDD.ipynb MapPartitionsRDD[2] at textFile at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "RDD.ipynb MapPartitionsRDD[2] at textFile at <console>:33"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd2 = sc1.textFile(\"RDD.ipynb\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.1.4 RDD 기본 액션\n",
    "\n",
    "#### 2.1.4.1 collect\n",
    "\n",
    "- RDD의 모든 원소를 모아서 배열로 돌려줍니다.\n",
    "- collect 연산을 수행하면 RDD에 있는 다른 모든 요소들이 collect 연산을 호출한 서버의 메모리에 수집되기 때문에 전체 데이터를 담을 수 있는 충분한 메모리 공간이 확보되었을 때 사용"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1, 2, 3, 4, 5, 6, 7, 8, 9, 10\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[9] at parallelize at <console>:36\n",
       "result = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc1.parallelize(1 to 10)\n",
    "val result = rdd.collect\n",
    "println(result.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.4.2 count\n",
    "\n",
    "* count는 RDD구성하는 전체 요소의 개수를 반환"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[10] at parallelize at <console>:36\n",
       "result = 10\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc1.parallelize(1 to 10)\n",
    "val result = rdd.count\n",
    "println(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.1.5 RDD transformation\n",
    "\n",
    "- 기존 RDD를 이용해 새로운 RDD를 생성하는 연산\n",
    "- 분류\n",
    "    - 맵 : 요소 간의 사상을 정의한 함수를 RDD에 속하는 모든 요소에 적용해 새로운 RDD 생성\n",
    "    - 그룹화 : 특정 조건에 따라 요소를 그룹화하거나 특정함수를 적용\n",
    "    - 집합 : RDD에 포함된 요소를 하나의 집합으로 간주할 때 서로 다른 RDD 간에 합집합, 교집합 등을 계산\n",
    "    - 파티션 : RDD의 파티션 개수를 조정\n",
    "    - 필터와 정렬 연산 : 특정 조건을 만족하는 요소만 선택하거나 각 요소를 정해진 기준에 따라 정렬\n",
    "- RDD가 제공하는 주요 연산이 RDD를 구성하고 있는 데이터 유형과 밀접한 관계를 맺는다"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [map 관련 연산]\n",
    "\n",
    "##### map\n",
    "map은 하나의 입력을 받아서 하나의 값을 돌려주는 함수를 인자로 받습니다.\n",
    "이 함수를 RDD에 속하는 모든 요소에 적용한 뒤 그 결과로 구성된 새로운 RDD를 생성해 반환합니다."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2, 3, 4, 5, 6, 7, 8, 9, 10, 11\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[13] at parallelize at <console>:36\n",
       "result = MapPartitionsRDD[14] at map at <console>:37\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[14] at map at <console>:37"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc1.parallelize(1 to 10)\n",
    "val result = rdd.map(_ + 1)\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** RDD는 그 자체로는 타입이 될 수 없고 반드시 타입 매개변수를 지정해서 정의해야 하는 매개변수화한 타입\n",
    "\n",
    "`map[U](f: (T) => U)(implicit arg0: ClassTag[U]): RDD[U]`\n",
    "\n",
    "`T타입을 U타입으로 변환하는 함수 f를 이용해 RDD<T> 타입의 RDD를 RDD<U> 타입으로 변환하는 메서드`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.2 flatMap\n",
    "\n",
    "- 각 배열 속에 포함된 요소를 모두 배열 밖으로 끄집어 내는 작어을 해야 할 경우\n",
    "- 하나의 입력값에 대응하는 반환값이 여러애리 경우"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{{apple, orange}, {grape, apple, mango}, {blueberry, tomato, orange}}\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "fruits = List(apple,orange, grape,apple,mango, blueberry,tomato,orange)\n",
       "rdd1 = ParallelCollectionRDD[17] at parallelize at <console>:40\n",
       "rdd2 = MapPartitionsRDD[18] at map at <console>:43\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[18] at map at <console>:43"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "// 단어 3개를 가진 List 생성\n",
    "val fruits = List(\"apple,orange\", \"grape,apple,mango\", \"blueberry,tomato,orange\")\n",
    "\n",
    "// RDD 생성\n",
    "val rdd1 = sc.parallelize(fruits)\n",
    "\n",
    "// RDD의 map() 메서드를 각 단어를 \",\"를 기준으로 분리\n",
    "val rdd2 = rdd1.map(_.split(\",\"))\n",
    "\n",
    "println(rdd2.collect().map(_.mkString(\"{\", \", \", \"}\")).mkString(\"{\", \", \", \"}\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{apple, orange, grape, apple, mango, blueberry, tomato, orange}\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "fruits = List(apple,orange, grape,apple,mango, blueberry,tomato,orange)\n",
       "rdd1 = ParallelCollectionRDD[19] at parallelize at <console>:37\n",
       "rdd2 = MapPartitionsRDD[20] at flatMap at <console>:38\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[20] at flatMap at <console>:38"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val fruits = List(\"apple,orange\", \"grape,apple,mango\", \"blueberry,tomato,orange\")\n",
    "val rdd1 = sc.parallelize(fruits)\n",
    "val rdd2 = rdd1.flatMap(_.split(\",\"))\n",
    "\n",
    "println(rdd2.collect().mkString(\"{\", \", \", \"}\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{0, 6}\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "fruits = List(apple,orange, grape,apple,mango, blueberry,tomato,orange)\n",
       "rdd1 = ParallelCollectionRDD[23] at parallelize at <console>:37\n",
       "rdd2 = MapPartitionsRDD[24] at flatMap at <console>:38\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[24] at flatMap at <console>:38"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val fruits = List(\"apple,orange\", \"grape,apple,mango\", \"blueberry,tomato,orange\")\n",
    "val rdd1 = sc.parallelize(fruits)\n",
    "val rdd2 = rdd1.flatMap(log => {\n",
    "    // apple이라는 단어가 포함된 경우만 처리하고 싶다.\n",
    "    if (log.contains(\"apple\")) {\n",
    "        Some(log.indexOf(\"apple\"))\n",
    "    } else {\n",
    "        None\n",
    "    }\n",
    "})\n",
    "println(rdd2.collect().mkString(\"{\", \", \", \"}\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### mapPartitions\n",
    "\n",
    "map, flatMap RDD의 각 요소를 하나씩 처리한다면\n",
    "mapPartitions()는 파티션 단위로 처리\n",
    "인자로 전달받은 함수를 파티션 단위로 적용하고 그 결과로 구성된 새로운 RDD를 생성하는 메서드\n",
    "\n",
    "파티션 단위로 파티션에 속한 모든 요소를 한번의 함수 호출로 처리할 수 있음\n",
    "\n",
    "데이터베이스 연결과 같은 고비용의 자원을 파티션 단위로 공유해 사용할 수 있는 장점"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2, 3, 4, 5, 6, 7, 8, 9, 10, 11\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[25] at parallelize at <console>:36\n",
       "rdd2 = MapPartitionsRDD[26] at mapPartitions at <console>:37\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[26] at mapPartitions at <console>:37"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(1 to 10, 3)\n",
    "val rdd2 = rdd1.mapPartitions(numbers => {\n",
    "    println(\"DB 연결\")\n",
    "    numbers.map{\n",
    "        number => number + 1\n",
    "    }\n",
    "})\n",
    "println(rdd2.collect().mkString( \", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.4 mapPartitionWithIndex\n",
    "\n",
    "인자로 전달받은 함수를 파티션 단위로 적용하고 그 결과값으로 구성된 새로운 RDDf르 생성하는 메서드\n",
    "\n",
    "mapPartition과 다른 점은 파티션의 인덱스 정보도 함께 전달해 준다는 점"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5, 6, 7\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[27] at parallelize at <console>:35\n",
       "rdd2 = MapPartitionsRDD[28] at mapPartitionsWithIndex at <console>:36\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[28] at mapPartitionsWithIndex at <console>:36"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(1 to 10, 3)\n",
    "val rdd2 = rdd1.mapPartitionsWithIndex((idx, numbers) => {\n",
    "    numbers.flatMap {\n",
    "        case number if idx == 1 => Option(number + 1)\n",
    "        case _                  => None\n",
    "    }\n",
    "})\n",
    "println(rdd2.collect().mkString( \", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### mapValues\n",
    "\n",
    "RDD의 요소가 키와 값이 쌍으로 이루고 있는 경우 PairRDD라는 용어를 사용\n",
    "\n",
    "- RDD의 모든 요소들이 키와 값의 쌍을 이루고 있는 경우에만 사용 가능한 메서드\n",
    "- 인자로 전달받은 함수를 \"값\"에 해당하는 요소에만 적용하고 그 결과로 구성된 새로운 RDD를 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,2)\t(b,2)\t(c,2)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = MapPartitionsRDD[30] at map at <console>:41\n",
       "result = MapPartitionsRDD[31] at mapValues at <console>:42\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[31] at mapValues at <console>:42"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "// (키,값) 쌍으로 구성된 RDD 생성\n",
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\")).map((_, 1))\n",
    "val result = rdd1.mapValues(_ + 1)\n",
    "println(result.collect.mkString(\"\\t\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.6 flatMapValues\n",
    "\n",
    "- RDD의 구성요소가 키와 값의 쌍으로 구성된 경우에만 적용\n",
    "- 키는 그대로 두고 값에 해당하는 요소만을 대상으로 flatMap() 연산을 적용"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(1,a)\t(1,b)\t(2,a)\t(2,c)\t(3,d)\t(3,e)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[34] at parallelize at <console>:35\n",
       "result = MapPartitionsRDD[35] at flatMapValues at <console>:36\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[35] at flatMapValues at <console>:36"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(Seq((1, \"a,b\"), (2, \"a,c\"), (3, \"d,e\")))\n",
    "val result = rdd.flatMapValues(_.split(\",\"))\n",
    "println(result.collect.mkString(\"\\t\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### [그룹과 관련된 연산들]\n",
    "\n",
    "#### 2.1.5.7 zip\n",
    "\n",
    "- 두개의 서로 다른 RDD를 각 요소의 인덱스에 따라 하나의 (키, 값) 쌍으로 묶어줍니다.\n",
    "- 첫번째 RDD의 n번째 요소를 키로하고 두번째 RDD의 n번째 요소를 값으로 하는 순서쌍을 생성\n",
    "- 이때 두 RDD는 같은 개수의 파티션을 가지고 있고 각 파티션에 속하는 요소의 수는 동일하다고 가정\n",
    "- 서로 크기가 다른 RDD 간에는 zip() 메서드를 사용할 수 없습니다."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,1), (b,2), (c,3)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[36] at parallelize at <console>:38\n",
       "rdd2 = ParallelCollectionRDD[37] at parallelize at <console>:39\n",
       "result = ZippedPartitionsRDD2[38] at zip at <console>:40\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ZippedPartitionsRDD2[38] at zip at <console>:40"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\"))\n",
    "val rdd2 = sc.parallelize(List(1,2,3))\n",
    "val result = rdd1.zip(rdd2)\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.8 zipPartitions\n",
    "\n",
    "- 파티션 단위로 zip() 연산을 수행하고 특정함수를 적용해 그 결과로 구성된 새로운 RDD를 생성\n",
    "- RDD의 파티션 개수만 동일하면 됨\n",
    "- zipPartitions는 최대 4개의 RDD 지정 가능\n",
    "- 병합에 사용할 함수를 인자로 전달받아서 사용할 수 있음"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "a1, b2, b3, c5, c6\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[42] at parallelize at <console>:36\n",
       "rdd2 = ParallelCollectionRDD[43] at parallelize at <console>:37\n",
       "result = ZippedPartitionsRDD2[44] at zipPartitions at <console>:38\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ZippedPartitionsRDD2[44] at zipPartitions at <console>:38"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\"), 3)\n",
    "val rdd2 = sc.parallelize(List(1,2,3,5,6), 3)\n",
    "val result = rdd1.zipPartitions(rdd2) {\n",
    "    (it1, it2) => for {\n",
    "        v1 <- it1\n",
    "        v2 <- it2\n",
    "    } yield v1 + v2\n",
    "}\n",
    "\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.9 groupBy\n",
    "\n",
    "- RDD의 요소를 일정한 기준에 따라 여러개의 그룹으로 나누고 이 그룹으로 구성된 새로운 RDD를 생성\n",
    "- 각 그룹은 키와 그 키에 속한 요소의 시퀀스로 구성\n",
    "- 메서드의 인자로 전달하는 함수가 각 그룹의 키를 결정하는 역할을 담당"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "even, [2,4,6,8,10]\n",
      "odd, [1,3,5,7,9]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[46] at parallelize at <console>:38\n",
       "result = ShuffledRDD[48] at groupBy at <console>:39\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ShuffledRDD[48] at groupBy at <console>:39"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 10)\n",
    "val result = rdd.groupBy{\n",
    "    case i: Int if (i % 2 == 0) => \"even\"\n",
    "    case i: Int if (i % 2 == 1) => \"odd\"\n",
    "    case _                      => \"\"\n",
    "}\n",
    "result.collect.foreach {\n",
    "    v => println(s\"\"\"${v._1}, [${v._2.mkString(\",\")}]\"\"\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.10 groupByKey\n",
    "\n",
    "- RDD의 구성요소가 키와 값으로 쌍으로 이루어진 경우에 사용 가능\n",
    "- 키를 기준으로 같은 키를 가진 요소들로 그룹을 만들고 이 그룹들로 구성된 새로운 RDD를 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "a, [1]\n",
      "b, [1,1]\n",
      "c, [1,1]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = MapPartitionsRDD[50] at map at <console>:35\n",
       "result = ShuffledRDD[51] at groupByKey at <console>:36\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ShuffledRDD[51] at groupByKey at <console>:36"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(\"a\", \"b\", \"c\", \"b\", \"c\")).map((_, 1))\n",
    "val result = rdd.groupByKey\n",
    "result.collect.foreach {\n",
    "    v => println(s\"\"\"${v._1}, [${v._2.mkString(\",\")}]\"\"\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.11 cogroup\n",
    "\n",
    "- RDD의 구성요소가 키와 값의 쌍으로 구성된 경우에만 사용할 수 있는 메서드\n",
    "- 여러 RDD에서 같은 키를 갖는 값 요소를 찾아서 키와 그 키에 속하는 요소의 시퀀스로 구성된 튜플을 만들고 그 튜플들로 구성된 새로운 RDD를 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(k3, [v3,v4], [])\n",
      "(k2, [v2], [])\n",
      "(k1, [v1], [v5])\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[56] at parallelize at <console>:36\n",
       "rdd2 = ParallelCollectionRDD[57] at parallelize at <console>:37\n",
       "result = MapPartitionsRDD[59] at cogroup at <console>:38\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[59] at cogroup at <console>:38"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List((\"k1\", \"v1\"), (\"k2\", \"v2\"), (\"k3\", \"v3\"), (\"k3\", \"v4\")))\n",
    "val rdd2 = sc.parallelize(List((\"k1\", \"v5\")))\n",
    "val result = rdd1.cogroup(rdd2)\n",
    "result.collect.foreach {\n",
    "    case (k, (v_1, v_2)) => {\n",
    "        println(s\"\"\"($k, [${v_1.mkString(\",\")}], [${v_2.mkString(\",\")}])\"\"\")\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.12 distint\n",
    "\n",
    "- RDD의 원소에서 중복을 제외한 요소로만 구성된 새로운 RDD를 생성하는 메서드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4, 1, 6, 3, 5, 2\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[60] at parallelize at <console>:38\n",
       "result = MapPartitionsRDD[63] at distinct at <console>:39\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[63] at distinct at <console>:39"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(1,2,3,4,5,5,6,2))\n",
    "val result = rdd.distinct\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.13 cartesian\n",
    "\n",
    "- 두 RDD 요소의 카테시안곱을 구하고 그 결과를 요소로 하는 새로운 RDD를 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,1), (a,2), (a,3), (b,1), (b,2), (b,3), (c,1), (c,2), (c,3)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[64] at parallelize at <console>:38\n",
       "rdd2 = ParallelCollectionRDD[65] at parallelize at <console>:39\n",
       "result = CartesianRDD[66] at cartesian at <console>:40\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "CartesianRDD[66] at cartesian at <console>:40"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\"))\n",
    "val rdd2 = sc.parallelize(List(1,2,3))\n",
    "val result = rdd1.cartesian(rdd2)\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 2.1.5.14 subtract\n",
    "\n",
    "- rdd1과 rdd2라는 두개의 RDD가 있을 때, rdd1.subtract(rdd2)는 rdd1에는 속하고 rdd2엔느 속하지 않는 요소로 구성된 새로운 RDD 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "b\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[0] at parallelize at <console>:27\n",
       "rdd2 = ParallelCollectionRDD[1] at parallelize at <console>:28\n",
       "result = MapPartitionsRDD[5] at subtract at <console>:29\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[5] at subtract at <console>:29"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\"))\n",
    "val rdd2 = sc.parallelize(List(\"a\",\"c\"))\n",
    "val result = rdd1.subtract(rdd2)\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.15  union\n",
    "\n",
    "- rdd1과 rdd2라는 두개의 RDD가 있을때 rdd1 또는 rdd2에 속하는 요소로 구성된 새로운 RDD를 생성하는 메서드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "a, b, c, d, e, f, a\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[6] at parallelize at <console>:31\n",
       "rdd2 = ParallelCollectionRDD[7] at parallelize at <console>:32\n",
       "result = UnionRDD[8] at union at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "UnionRDD[8] at union at <console>:33"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\"))\n",
    "val rdd2 = sc.parallelize(List(\"d\", \"e\", \"f\", \"a\"))\n",
    "val result = rdd1.union(rdd2)\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.16 intersection\n",
    "\n",
    "- rdd1과 rdd2라는 두개의 RDD가 있을 때 rdd1과 rdd2에 동시에 속하는 요소로 구성된 새로운 RDD를 생성하는 메서드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "a, e\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[33] at parallelize at <console>:31\n",
       "rdd2 = ParallelCollectionRDD[34] at parallelize at <console>:32\n",
       "result = MapPartitionsRDD[40] at intersection at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[40] at intersection at <console>:33"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\",\"e\",\"a\", \"b\"))\n",
    "val rdd2 = sc.parallelize(List(\"d\", \"e\", \"f\", \"a\"))\n",
    "val result = rdd1.intersection(rdd2)\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.17 join\n",
    "\n",
    "- RDD 의 구성요소가 키와 값의 쌍으로 구성된 경우에 사용할 수 있는 메서드 입니다.\n",
    "- 두 RDD에서 같은 키를 가지고 있는 요소를 모아서 그룹을 형성하고, 그 결과로 구성된 새로운 RDD를 생성하는 메서드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,(1,2))\n",
      "(a,(1,2))\n",
      "(e,(1,2))\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = MapPartitionsRDD[42] at map at <console>:31\n",
       "rdd2 = MapPartitionsRDD[44] at map at <console>:32\n",
       "result = MapPartitionsRDD[47] at join at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[47] at join at <console>:33"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\",\"e\",\"a\", \"b\")).map((_, 1))\n",
    "val rdd2 = sc.parallelize(List(\"d\", \"e\", \"f\", \"a\")).map((_, 2))\n",
    "val result = rdd1.join(rdd2)\n",
    "println(result.collect.mkString(\"\\n\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.18 leftOIuterJoin, rightOuterJoin"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,(1,Some(2)))\n",
      "(a,(1,Some(2)))\n",
      "(b,(1,None))\n",
      "(b,(1,None))\n",
      "(c,(1,None))\n",
      "(e,(1,Some(2)))\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = MapPartitionsRDD[49] at map at <console>:31\n",
       "rdd2 = MapPartitionsRDD[51] at map at <console>:32\n",
       "result = MapPartitionsRDD[54] at leftOuterJoin at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[54] at leftOuterJoin at <console>:33"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\",\"e\",\"a\", \"b\")).map((_, 1))\n",
    "val rdd2 = sc.parallelize(List(\"d\", \"e\", \"f\", \"a\")).map((_, 2))\n",
    "val result = rdd1.leftOuterJoin(rdd2)\n",
    "println(result.collect.mkString(\"\\n\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,(Some(1),2))\n",
      "(a,(Some(1),2))\n",
      "(d,(None,2))\n",
      "(e,(Some(1),2))\n",
      "(f,(None,2))\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = MapPartitionsRDD[56] at map at <console>:31\n",
       "rdd2 = MapPartitionsRDD[58] at map at <console>:32\n",
       "result = MapPartitionsRDD[61] at rightOuterJoin at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[61] at rightOuterJoin at <console>:33"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\",\"e\",\"a\", \"b\")).map((_, 1))\n",
    "val rdd2 = sc.parallelize(List(\"d\", \"e\", \"f\", \"a\")).map((_, 2))\n",
    "val result = rdd1.rightOuterJoin(rdd2)\n",
    "println(result.collect.mkString(\"\\n\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.19 subtractByKey\n",
    "\n",
    "- RDD의 구성요소가 키와 값의 쌍으로 구성된 경우\n",
    "- rdd1과 rdd2라는 두 RDD가 있을 때 rdd1.subtractByKey(rdd2)는 rdd1의 요소중에서 rdd2에 같은 키가 존재하는 요소를 제외"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(b,1)\n",
      "(b,1)\n",
      "(c,1)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = MapPartitionsRDD[63] at map at <console>:31\n",
       "rdd2 = MapPartitionsRDD[65] at map at <console>:32\n",
       "result = SubtractedRDD[66] at subtractByKey at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "SubtractedRDD[66] at subtractByKey at <console>:33"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"a\", \"b\", \"c\",\"e\",\"a\", \"b\")).map((_, 1))\n",
    "val rdd2 = sc.parallelize(List(\"d\", \"e\", \"f\", \"a\")).map((_, 2))\n",
    "val result = rdd1.subtractByKey(rdd2)\n",
    "println(result.collect.mkString(\"\\n\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### [집계와 관련된 연산들]\n",
    "\n",
    "#### 2.1.5.20 reduceByKey\n",
    "\n",
    "- RDD의 구성요소가 키와 값의 쌍으로 구성된 경우\n",
    "- 같은 키를 가진 값들을 하나로 병합해 키-값 쌍으로 구성된 새로운 RDD를 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,2),(b,1)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = MapPartitionsRDD[68] at map at <console>:31\n",
       "result = ShuffledRDD[69] at reduceByKey at <console>:32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ShuffledRDD[69] at reduceByKey at <console>:32"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(\"a\", \"b\", \"a\")).map((_, 1))\n",
    "val result = rdd.reduceByKey(_ + _)\n",
    "println(result.collect.mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.21 foldByKey\n",
    "\n",
    "- RDD의 구성요소가 키와 값의 쌍으로 구성된 경우\n",
    "- 전반적인 동작은 reduceByKey 유사\n",
    "- 병합 연산의 초기 값을 메서더의 인자로 전달해서 병합시 사용할 수 있다는 점에서 차이"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,4),(b,2)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = MapPartitionsRDD[74] at map at <console>:30\n",
       "result = ShuffledRDD[75] at foldByKey at <console>:31\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ShuffledRDD[75] at foldByKey at <console>:31"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(\"a\", \"b\", \"a\")).map((_, 1))\n",
    "val result = rdd.foldByKey(1)(_ + _)\n",
    "println(result.collect.mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.22 combineByKey\n",
    "\n",
    "- RDD의 구성요소가 키와 값의 쌍으로 구성된 경우\n",
    "- 같은 키를 가진 값들을 하나로 병합하는 기능을 수행하지만 병합을 수행하는 과정에서 값의 타입이 바뀔 수 있다.\n",
    "    - createCombiner : 값을 병합하기 위한 콤바이너\n",
    "    - mergeValue : 키에 대한 콤바이너가 이미 존재한담녀 새로운 콤바이너를 만들지 않고 병합\n",
    "    - mergeCombiner : \n",
    "        - createCombiner(), mergeValue()는 파티션 단위\n",
    "        - mergeCombiner는 병합이 끝난 콤바이너들끼리 다시 병합"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(A,avg: 75)\n",
      "(B,avg: 76)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "defined class Record\n",
       "data = List((A,100), (B,80), (A,50), (B,60), (B,90))\n",
       "rdd = ParallelCollectionRDD[84] at parallelize at <console>:39\n",
       "createCombiner = > Record = <function1>\n",
       "mergeValue = > Record = <function2>\n",
       "mergeCombiners = > Record = <function2>\n",
       "result = ShuffledRDD[85] at combineByKey at <console>:43\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "lastException: Throwable = null\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ShuffledRDD[85] at combineByKey at <console>:43"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "case class Record(var amount: Long, var number: Long = 1) {\n",
    "    def map(v: Long) = Record(v)\n",
    "    def add(amount: Long): Record = {\n",
    "        add(map(amount))\n",
    "    }\n",
    "    def add(other: Record): Record = {\n",
    "        this.number += other.number\n",
    "        this.amount += other.amount\n",
    "        this\n",
    "    }\n",
    "    override def toString: String = s\"avg: ${amount / number}\"\n",
    "}\n",
    "\n",
    "// combineByKey()를 이용한 평균값 계산\n",
    "val data = Seq((\"A\", 100L), (\"B\", 80L), (\"A\", 50L), (\"B\", 60L), (\"B\", 90L))\n",
    "val rdd = sc.parallelize(data)\n",
    "val createCombiner = (v: Long) => Record(v)\n",
    "val mergeValue = (c: Record, v: Long) => c.add(v)\n",
    "val mergeCombiners = (c1: Record, c2: Record) => c1.add(c2)\n",
    "val result = rdd.combineByKey(createCombiner, mergeValue, mergeCombiners)\n",
    "println(result.collect.mkString(\"\\n\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.23 aggregateByKey\n",
    "\n",
    "- RDD의 구성요소가 키와 값의 쌍으로 구성된 경우\n",
    "- 병합을 시작할 때 초깃값을 생성하는 하는 부분을 제외하면 combineByKey와 동일한 동작을 수행"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(A,avg: 75)\n",
      "(B,avg: 80)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "Name: java.lang.ArithmeticException\n",
       "Message: / by zero\n",
       "StackTrace:   at Record.toString(<console>:34)\n",
       "  at scala.runtime.ScalaRunTime$.scala$runtime$ScalaRunTime$$inner$1(ScalaRunTime.scala:332)\n",
       "  at scala.runtime.ScalaRunTime$.stringOf(ScalaRunTime.scala:337)\n",
       "  at scala.runtime.ScalaRunTime$.replStringOf(ScalaRunTime.scala:345)\n",
       "  at .$print$lzycompute(<console>:12)\n",
       "  at .$print(<console>:6)\n",
       "  at $print(<console>)\n",
       "  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n",
       "  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n",
       "  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
       "  at java.lang.reflect.Method.invoke(Method.java:498)\n",
       "  at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:786)\n",
       "  at scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:1047)\n",
       "  at scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:638)\n",
       "  at scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:637)\n",
       "  at scala.reflect.internal.util.ScalaClassLoader$class.asContext(ScalaClassLoader.scala:31)\n",
       "  at scala.reflect.internal.util.AbstractFileClassLoader.asContext(AbstractFileClassLoader.scala:19)\n",
       "  at scala.tools.nsc.interpreter.IMain$WrappedRequest.loadAndRunReq(IMain.scala:637)\n",
       "  at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:569)\n",
       "  at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:565)\n",
       "  at org.apache.toree.kernel.interpreter.scala.ScalaInterpreterSpecific$$anonfun$interpretAddTask$1$$anonfun$apply$3.apply(ScalaInterpreterSpecific.scala:385)\n",
       "  at org.apache.toree.kernel.interpreter.scala.ScalaInterpreterSpecific$$anonfun$interpretAddTask$1$$anonfun$apply$3.apply(ScalaInterpreterSpecific.scala:380)\n",
       "  at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)\n",
       "  at scala.Console$.withErr(Console.scala:80)\n",
       "  at org.apache.toree.global.StreamState$$anonfun$1$$anonfun$apply$1.apply(StreamState.scala:73)\n",
       "  at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)\n",
       "  at scala.Console$.withOut(Console.scala:53)\n",
       "  at org.apache.toree.global.StreamState$$anonfun$1.apply(StreamState.scala:72)\n",
       "  at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)\n",
       "  at scala.Console$.withIn(Console.scala:124)\n",
       "  at org.apache.toree.global.StreamState$.withStreams(StreamState.scala:71)\n",
       "  at org.apache.toree.kernel.interpreter.scala.ScalaInterpreterSpecific$$anonfun$interpretAddTask$1.apply(ScalaInterpreterSpecific.scala:379)\n",
       "  at org.apache.toree.kernel.interpreter.scala.ScalaInterpreterSpecific$$anonfun$interpretAddTask$1.apply(ScalaInterpreterSpecific.scala:379)\n",
       "  at org.apache.toree.utils.TaskManager$$anonfun$add$2$$anon$2.run(TaskManager.scala:134)\n",
       "  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\n",
       "  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "// combineByKey()를 이용한 평균값 계산\n",
    "val data = Seq((\"A\", 100L), (\"B\", 80L), (\"A\", 50L), (\"B\", 70L), (\"B\", 90L))\n",
    "val rdd = sc.parallelize(data)\n",
    "\n",
    "val zero = Record(0, 0)\n",
    "val mergeValue = (c: Record, v: Long) => c.add(v)\n",
    "val mergeCombiners = (c1: Record, c2: Record) => c1.add(c2)\n",
    "\n",
    "val result = rdd.aggregateByKey(zero)(mergeValue, mergeCombiners)\n",
    "println(result.collect.mkString(\"\\n\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### pipe 및 파티션과 관련된 연산\n",
    "\n",
    "#### 2.1.5.24 pipe\n",
    "\n",
    "- pipe를 이용하면 데이터를 처리하는 과정에서 외부 프로세스를 활용할 수 있음"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1,3, 4,6, 7,9\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[92] at parallelize at <console>:36\n",
       "result = PipedRDD[93] at pipe at <console>:37\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "lastException: Throwable = null\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "PipedRDD[93] at pipe at <console>:37"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(\"1,2,3\", \"4,5,6\", \"7,8,9\"))\n",
    "val result = rdd.pipe(\"cut -f 1,3 -d ,\")\n",
    "println(result.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.25 coalesce 와 repartition\n",
    "\n",
    "- 현재 RDD의 파티션 개수를 조정\n",
    "- repartion()\n",
    "    - 파티션 수를 줄이거나 늘이거나 가능\n",
    "    - 셔플기반으로 동작 수행\n",
    "- coalesce()\n",
    "    - 파티션을 줄이는 것만 가능\n",
    "    - 셔플을 사용하지 않고"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "partition size: 10\n",
      "partition size: 5\n",
      "partition size: 10\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = ParallelCollectionRDD[94] at parallelize at <console>:30\n",
       "rdd2 = CoalescedRDD[95] at coalesce at <console>:31\n",
       "rdd3 = MapPartitionsRDD[99] at repartition at <console>:32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[99] at repartition at <console>:32"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(1 to 1000000, 10)\n",
    "val rdd2 = rdd1.coalesce(5)\n",
    "val rdd3 = rdd2.repartition(10)\n",
    "\n",
    "println(s\"partition size: ${rdd1.getNumPartitions}\")\n",
    "println(s\"partition size: ${rdd2.getNumPartitions}\")\n",
    "println(s\"partition size: ${rdd3.getNumPartitions}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.26 repartionAndSortWithinPartitions\n",
    "\n",
    "- 같은 성격을 지닌 데이터를 같은 파티션으로 분리하고 싶을 때\n",
    "- RDD를 구성하는 모든 데이터를 특정 기준에 따라서 여러 개의 파티션으로 분리하고 각 파티션 단위로 정렬을 수행한 뒤 이 결과로 새로운 RDD를 생성\n",
    "- 키와 값 쌍"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "r = scala.util.Random$@13a021fc\n",
       "data = Vector((88,-), (16,-), (17,-), (75,-), (90,-), (41,-), (76,-), (8,-), (75,-), (94,-))\n",
       "rdd1 = ParallelCollectionRDD[102] at parallelize at <console>:37\n",
       "rdd2 = ShuffledRDD[103] at repartitionAndSortWithinPartitions at <console>:38\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ShuffledRDD[103] at repartitionAndSortWithinPartitions at <console>:38"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.HashPartitioner\n",
    "\n",
    "val r = scala.util.Random\n",
    "val data = for (i <- 1 to 10) yield (r.nextInt(100), \"-\")\n",
    "val rdd1 = sc.parallelize(data)\n",
    "val rdd2 = rdd1.repartitionAndSortWithinPartitions(new HashPartitioner(3))\n",
    "//결과 검중\n",
    "rdd2.foreachPartition(it => {\n",
    "    println(\"==========\")\n",
    "    it.foreach(v => println(v))\n",
    "})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.27 partitionBy\n",
    "\n",
    "- 키와 값의 쌍\n",
    "- org.apache.spark.Partitioner 클래스의 인자 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rdd1: 5, rdd2: 2\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = MapPartitionsRDD[108] at map at <console>:32\n",
       "rdd2 = ShuffledRDD[109] at partitionBy at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ShuffledRDD[109] at partitionBy at <console>:33"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"apple\", \"mouse\", \"monitor\"), 5).map{a => (a, a.length)}\n",
    "val rdd2 = rdd1.partitionBy(new HashPartitioner(2))\n",
    "println(s\"rdd1: ${rdd1.getNumPartitions}, rdd2: ${rdd2.getNumPartitions}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  필터와 정렬 연산\n",
    "\n",
    "#### 2.1.5.28 filter\n",
    "\n",
    "- 원하는 요소만 남기고 원하지 않는 요소는 걸러내는 동작"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3,4,5\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[110] at parallelize at <console>:32\n",
       "result = MapPartitionsRDD[111] at filter at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[111] at filter at <console>:33"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 5)\n",
    "val result = rdd.filter(_ > 2)\n",
    "println(result.collect.mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.29 sortByKey\n",
    "\n",
    "- 키 값을 기준으로 요소를 정렬"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,1),(q,1),(z,1)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[112] at parallelize at <console>:32\n",
       "result = ShuffledRDD[116] at sortByKey at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ShuffledRDD[116] at sortByKey at <console>:33"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(\"q\", \"z\", \"a\"))\n",
    "val result = rdd.map((_, 1)).sortByKey()\n",
    "println(result.collect.mkString(\",\"))    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.30 keys, values\n",
    "\n",
    "- 키와 값의 쌍\n",
    "- keys()\n",
    "    - 키에 해당하는 요소로 구성된 RDD 생성\n",
    "- values()\n",
    "    - 값에 해당하는 요소로 구성된 RDD 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "apple,mouse,monitor\n",
      "5,5,7\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd1 = MapPartitionsRDD[118] at map at <console>:31\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[118] at map at <console>:31"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd1 = sc.parallelize(List(\"apple\", \"mouse\", \"monitor\"), 5).map{a => (a, a.length)}\n",
    "println(rdd1.keys.collect.mkString(\",\"))\n",
    "println(rdd1.values.collect.mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.5.31 sample\n",
    "\n",
    "- 샘플을 추출해 새로운 RDD 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3,6,9,13,14\n",
      "1,3,4,4,4\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[124] at parallelize at <console>:33\n",
       "result1 = PartitionwiseSampledRDD[125] at sample at <console>:34\n",
       "result2 = PartitionwiseSampledRDD[126] at sample at <console>:35\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "PartitionwiseSampledRDD[126] at sample at <console>:35"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 100)\n",
    "val result1 = rdd.sample(false, 0.5)\n",
    "val result2 = rdd.sample(true, 1.5)\n",
    "println(result1.take(5).mkString(\",\"))\n",
    "println(result2.take(5).mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.1.6 RDD 액션\n",
    "- lazy eveluation\n",
    "- 계산에 필요한 정보를 누적해서 내포하고 있다가 실제로 계산이 필요한 시점이 되어서야 실행\n",
    "\n",
    "### 출력과 관련된 연산\n",
    "\n",
    "#### 2.1.6.1 first\n",
    "- 첫번째 요소 하나"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[127] at parallelize at <console>:32\n",
       "result = 5\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "5"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(5, 4, 1))\n",
    "val result = rdd.first\n",
    "println(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.2 take\n",
    "\n",
    "- 첫번째 요소로부터 순서대로 n개를 추출\n",
    "- 배열/리스트와 같은 컬렉션 타입 반환"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1,2,3,4,5\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[128] at parallelize at <console>:32\n",
       "result = Array(1, 2, 3, 4, 5)\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[1, 2, 3, 4, 5]"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 20, 5)\n",
    "val result = rdd.take(5)\n",
    "println(result.mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.3 takeSmple\n",
    "- 지정된 크기의 샘플을 추출\n",
    "\n",
    "#### 2.1.6.4 collect, count\n",
    "- collect()\n",
    "    - 모든 원소 모음\n",
    "- count()\n",
    "    - 원소의 갯수\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.5 countByValue\n",
    "- RDD에 속하는 각 값들이 나타나는 횟수를 구해서 맵 행태로 돌려주는 메서드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Map(1 -> 2, 2 -> 1, 3 -> 2)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[129] at parallelize at <console>:32\n",
       "result = Map(1 -> 2, 2 -> 1, 3 -> 2)\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Map(1 -> 2, 2 -> 1, 3 -> 2)"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(1,1,2,3,3))\n",
    "val result = rdd.countByValue\n",
    "println(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.6 reduce\n",
    "\n",
    "- RDD에 포함된 임의의 값 두개를 하나로 합치는 함수를 이용해 RDD에 포함된 모든 요소를 하나의 값으로 병합하고 반환하는 메서드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "55\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[133] at parallelize at <console>:32\n",
       "result = 55\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "55"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 10)\n",
    "val result = rdd.reduce(_ + _)\n",
    "println(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.7 fold\n",
    "\n",
    "- 같은 RDD내의 모든 요소를 대상으로 교환법칙과 결합법칙이 성립되는 바이너리 함수를 순차적용해 최종 결과를 구하는 메서드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "55\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[134] at parallelize at <console>:32\n",
       "result = 55\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "55"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 10)\n",
    "val result = rdd.fold(0)(_ + _)\n",
    "println(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.8 aggregate\n",
    "\n",
    "- reduce(), fold()는 모두 입력/출력 타입이 동일해야 한다는 제약이 있음\n",
    "- aggrecate()메서드는 제약이 없음\n",
    "- 3개의 인자\n",
    "    - zeroValue: 초기값\n",
    "    - seqOp : 파티션 내에서 병합\n",
    "    - combOp : 초종 병합"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "avg: 88\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[139] at parallelize at <console>:34\n",
       "result2 = avg: 88\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "lastException: Throwable = null\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "avg: 88"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(100, 80, 75, 90, 95), 3)\n",
    "val zeroValue = Record(0, 0)\n",
    "val seqOp = (r: Record, v: Int) => r.add(v)\n",
    "val combOp = (r1: Record, r2: Record) => r1.add(r2)\n",
    "val result1 = rdd.aggregate(zeroValue)(seqOp, combOp)\n",
    "println(result1.amount/result1.number)\n",
    "\n",
    "// 좀 더 간결\n",
    "val result2 = rdd.aggregate(Record(0, 0))(_ add _, _ add _)\n",
    "println(result2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.9 sum\n",
    "\n",
    "- 모든 요소가 double, Long등 숫자 타입일 경우"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "55.0\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[140] at parallelize at <console>:32\n",
       "result = 55.0\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "55.0"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 10)\n",
    "val result = rdd.sum\n",
    "println(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.10 foreach, foreachPartition\n",
    "\n",
    "- foreach()\n",
    "    - RDD의 모든 요소에 특정함수를 적용하는 메서드\n",
    "- foreachPartition()\n",
    "    - 파티션 단위로 적용\n",
    "    - 실행말 할뿐 리턴하지 않는다.\n",
    "- 인자로 전달받은 함수가 개별노드에서 실행"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[143] at parallelize at <console>:31\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[143] at parallelize at <console>:31"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 10, 3)\n",
    "rdd.foreach{ v => \n",
    "    println(s\"Value Side Effect: ${v}\")\n",
    "}\n",
    "\n",
    "rdd.foreachPartition(values => {\n",
    "    println(\"Partition Side Effect\")\n",
    "    for(v <- values) println(s\"Value Side Effect: ${v}\")\n",
    "})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.11 toDebugString\n",
    "\n",
    "- 디버깅을 위한 메서드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) CoalescedRDD[147] at coalesce at <console>:31 []\n",
      " |  MapPartitionsRDD[146] at map at <console>:31 []\n",
      " |  MapPartitionsRDD[145] at map at <console>:31 []\n",
      " |  ParallelCollectionRDD[144] at parallelize at <console>:31 []\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = CoalescedRDD[147] at coalesce at <console>:31\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "CoalescedRDD[147] at coalesce at <console>:31"
      ]
     },
     "execution_count": 58,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 100, 10).map(_ * 2).persist.map(_ + 1).coalesce(2)\n",
    "println(rdd.toDebugString)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.12 cache, persist, unpersist\n",
    "\n",
    "- 액션 연산이 수행될 때마다 관련 트랜스포메이션 연산을 반복\n",
    "- 기존에 사용했던 데이터가 메모리에 남아 있다면 그 데이터를 사용하지만 다른 이유로 인해 데이터 남아있지 않다면 RDD 생성 히스토리(리니지)를 이용해 복구 수행\n",
    "- cache와 persist는 첫 액션을 실행한 후에 RDD정보를 메모리 또는 디스크에 저장"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[148] at parallelize at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[148] at parallelize at <console>:33"
      ]
     },
     "execution_count": 59,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.storage.StorageLevel\n",
    "\n",
    "val rdd = sc.parallelize(1 to 100, 10)\n",
    "rdd.cache\n",
    "rdd.persist(StorageLevel.MEMORY_ONLY)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.6.13 partitions\n",
    "\n",
    "- RDD의 파티션 정보가 담긴 배열을 리턴"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10\n",
      "10\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[149] at parallelize at <console>:32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[149] at parallelize at <console>:32"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 100, 10)\n",
    "println(rdd.partitions.size)\n",
    "println(rdd.getNumPartitions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.1.7 RDD 데이터 불러오기와 저장하기\n",
    "\n",
    "- 다양한 포맷 지원"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.7.1 테스트 파일"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = RDD.ipynb MapPartitionsRDD[151] at textFile at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "RDD.ipynb MapPartitionsRDD[151] at textFile at <console>:30"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.textFile(\"RDD.ipynb\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[152] at parallelize at <console>:32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[152] at parallelize at <console>:32"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 1000, 3)\n",
    "rdd.saveAsTextFile(\"./sub1\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.7.2 오프젝트 파일(object file)\n",
    "\n",
    "- object file을 읽고 쓰는 기능 제공"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "126,127,128,129,130,131,132,133,134,135\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[154] at parallelize at <console>:34\n",
       "rdd1 = MapPartitionsRDD[158] at objectFile at <console>:36\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[158] at objectFile at <console>:36"
      ]
     },
     "execution_count": 64,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 1000)\n",
    "rdd.saveAsObjectFile(\"./sub_path\")\n",
    "val rdd1 = sc.objectFile[Int](\"./sub_path\")\n",
    "println(rdd1.take(10).mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.7.3 시퀀스 파일\n",
    "\n",
    "- Sequence file은 키와 값으로 구성된 데이터를 저장하는 Binary 파일 포맷"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,1), (b,1), (c,1), (b,1), (c,1)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = MapPartitionsRDD[160] at map at <console>:35\n",
       "rdd2 = MapPartitionsRDD[163] at sequenceFile at <console>:37\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[163] at sequenceFile at <console>:37"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(List(\"a\", \"b\", \"c\", \"b\", \"c\")).map((_, 1))\n",
    "rdd.saveAsSequenceFile(\"data/smaple/saveAsSeqFile/scala\")\n",
    "val rdd2 = sc.sequenceFile[String, Int](\"data/smaple/saveAsSeqFile/scala\")\n",
    "println(rdd2.collect.mkString(\", \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.1.8 클러스터 환경에서의 공유 변수\n",
    "\n",
    "#### broadcast variable\n",
    "- 스파크 잡이 실행되는 동안 클러스터 내의 모든 서버에서 공유할 수 있는 일기 전용 자원 설정"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "u1,u2\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "broadcateUsers = Broadcast(93)\n",
       "rdd = ParallelCollectionRDD[168] at parallelize at <console>:35\n",
       "result = MapPartitionsRDD[169] at filter at <console>:36\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[169] at filter at <console>:36"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val broadcateUsers = sc.broadcast(Set(\"u1\", \"u2\"))\n",
    "val rdd = sc.parallelize(List(\"u1\", \"u2\", \"u3\", \"u4\", \"u5\", \"u6\"), 3)\n",
    "val result = rdd.filter(broadcateUsers.value.contains(_))\n",
    "println(result.collect.mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Accumulator\n",
    "- 클러스터 내의 모든 서버가 공유하는 쓰기 공간을 제공함으로써 각 서버에서 발생하는 특정 이벤트의 수를 세거나 관찰하고 싶은 정보를 모아두는 등의 용도로 편리하게 활용"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "잘못된 데이터 수: 1\n",
      "잘못된 데이터: [u3]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "acc1 = LongAccumulator(id: 16202, name: Some(invlidFormat), value: 1)\n",
       "acc2 = CollectionAccumulator(id: 16203, name: Some(invalidFormat2), value: [u3])\n",
       "data = List(u1:addr1, u2:add2, u3, u4:addr4)\n",
       "rdd = ParallelCollectionRDD[173] at parallelize at <console>:38\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[173] at parallelize at <console>:38"
      ]
     },
     "execution_count": 72,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val acc1 = sc.longAccumulator(\"invlidFormat\")\n",
    "val acc2 = sc.collectionAccumulator[String](\"invalidFormat2\")\n",
    "val data = List(\"u1:addr1\", \"u2:add2\", \"u3\", \"u4:addr4\")\n",
    "val rdd = sc.parallelize(data)\n",
    "rdd.foreach{ v => \n",
    "    if (v.split(\":\").length != 2) {\n",
    "        acc1.add(1L)\n",
    "        acc2.add(v)\n",
    "    }\n",
    "}\n",
    "println(\"잘못된 데이터 수: \" + acc1.value)\n",
    "println(\"잘못된 데이터: \" + acc2.value)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 정리\n",
    "\n",
    "- RDD는 스파크에 다루는 데이터에 대한 추상 모델로서 메모리를 기반으로 동작하면서도 데이터를 처리하는 과정에서 데이터 누락되거나 유실되지 않게 하는 에러 복구 메커니즘이 있음\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "bespoke_spark - Scala",
   "language": "scala",
   "name": "bespoke_spark_scala"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "2.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}