1小时入门Spark之RDD编程.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 〇，编程环境"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "以下过程为Mac系统上单机版Spark练习编程环境的配置方法。\n",
    "注意：仅配置练习环境无需安装hadoop,无需安装scala.\n",
    "\n",
    "\n",
    "1，安装Java8\n",
    "\n",
    "https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html\n",
    "\n",
    "注意避免安装其它版本的jdk否则可能会有不兼容spark的情况\n",
    "\n",
    "\n",
    "2，下载spark并解压\n",
    "\n",
    "http://spark.apache.org/downloads.html\n",
    "\n",
    "解压到以下路径：\n",
    "\n",
    "Users/liangyun/ProgramFiles/spark-2.4.3-bin-hadoop2.7\n",
    "\n",
    "\n",
    "3，配置spark环境\n",
    "\n",
    "vim ~/.bashrc\n",
    "\n",
    "插入下面两条语句\n",
    "\n",
    "export SPARK_HOME=/Users/liangyun/ProgramFiles/spark-2.4.3-bin-hadoop2.7\n",
    "\n",
    "export PATH=$PATH:$SPARK_HOME/bin\n",
    "\n",
    "\n",
    "4，配置jupyter支持\n",
    "\n",
    "若未有安装jupyter可以下载Anaconda安装之。\n",
    "pip install toree\n",
    "\n",
    "jupyter toree install --spark_home=your-spark-home\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 一，运行Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark主要通过以下一些方式运行。\n",
    "\n",
    "1，通过spark-shell进入spark交互式环境,使用Scala语言。\n",
    "\n",
    "2，通过spark-submit提交Spark应用程序进行批处理。\n",
    "这种方式可以提交Scala或Java语言编写后生成的jar包或者Python脚本。\n",
    "\n",
    "3，通过pyspark进入pyspark交互式环境，使用Python语言。\n",
    "这种方式可以指定jupyter或者ipython为交互环境。\n",
    "\n",
    "4，通过zepplin notebook交互式执行。\n",
    "zepplin是jupyter notebook的apache对应产品。\n",
    "\n",
    "5, 安装Apache Toree - Scala内核。\n",
    "可以在jupyter 中运行spark-shell。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "使用spark-shell运行时，还可以添加常用的两个参数。\n",
    "\n",
    "一个是-master指定使用何种分布类型。\n",
    "\n",
    "第二个是-jars指定依赖的jar包。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "//local本地模式运行，默认使用4个逻辑CPU内核\n",
    "spark-shell\n",
    "\n",
    "//local本地模式运行，使用全部内核，添加 code.jar到classpath\n",
    "spark-shell  --master local[*] --jars code.jar \n",
    "\n",
    "//local本地模式运行，使用4个内核\n",
    "spark-shell  --master local[4]\n",
    "\n",
    "//standalone模式连接集群，指定url和端口号\n",
    "spark-shell  --master spark://master:7077\n",
    "\n",
    "//客户端模式连接YARN集群，Driver运行在本地，方便查看日志，调试时推荐使用。\n",
    "spark-shell  --master yarn-client\n",
    "\n",
    "//集群模式连接YARN集群，Driver运行在集群，本地机器计算和通信压力小，批量任务时推荐使用。\n",
    "spark-shell  --master yarn-cluster\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "//提交scala写的任务\n",
    "./bin/spark-submit --class org.apache.spark.examples.SparkPi \\\n",
    " --master yarn \\\n",
    " --deploy-mode cluster \\\n",
    " --driver-memory 4g \\\n",
    " --executor-memory 2g \\\n",
    " --executor-cores 1 \\\n",
    " --queue thequeue \\\n",
    " examples/jars/spark-examples*.jar 10 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "//提交python写的任务\n",
    "spark-submit --master yarn \\\n",
    "--executor-memory 6G \\\n",
    "--driver-memory 6G \\\n",
    "--deploy-mode cluster \\\n",
    "--num-executors 600 \\\n",
    "--conf spark.yarn.maxAppAttempts=1 \\\n",
    "--executor-cores 1 \\\n",
    "--conf spark.default.parallelism=2000 \\\n",
    "--conf spark.task.maxFailures=10 \\\n",
    "--conf spark.stage.maxConsecutiveAttempts=10 \\\n",
    "test.py "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 一，创建RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "创建RDD主要有两种方式，一个是textFile加载本地或者集群文件系统中的数据，\n",
    "\n",
    "第二个是用parallelize方法将Driver中的数据结构并行化成RDD。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "2\n",
      "3\n",
      "4\n",
      "5\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "file = file:///Users/liangyun/CodeFiles/spark_tutorial/data.txt\n",
       "rdd = file:///Users/liangyun/CodeFiles/spark_tutorial/data.txt MapPartitionsRDD[1] at textFile at <console>:29\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "file:///Users/liangyun/CodeFiles/spark_tutorial/data.txt MapPartitionsRDD[1] at textFile at <console>:29"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(0,0)\n",
      "(0,0)\n",
      "(0,0)\n",
      "(14,1)\n",
      "(7,1)\n",
      "(29,2)\n",
      "(15,2)\n",
      "(45,3)\n",
      "(24,3)\n",
      "(62,4)\n",
      "(34,4)\n",
      "(80,5)\n",
      "(45,5)\n",
      "(99,6)\n",
      "(57,6)\n",
      "(1,1)\n",
      "(3,2)\n",
      "(6,3)\n",
      "(10,4)\n",
      "(15,5)\n",
      "((0,0),(70,7))\n",
      "((70,7),(119,7))\n",
      "((189,14),(21,6))\n"
     ]
    }
   ],
   "source": [
    "//从本地文件系统中加载数据\n",
    "val file = \"file:///Users/liangyun/CodeFiles/spark_tutorial/data.txt\"\n",
    "val rdd = sc.textFile(file,3)\n",
    "rdd.collect.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "//从集群文件系统中加载数据\n",
    "val file = \"hdfs://localhost:9000/user/hadoop/data.txt\"\n",
    "//可以省去hdfs://localhost:9000\n",
    "val rdd = sc.textFile(file,3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[55] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//parallelize将Driver中的数据结构生成RDD,第二个参数指定分区数\n",
    "val rdd = sc.parallelize(1 to 10,2)\n",
    "rdd.collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[56] at makeRDD at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//makeRDD的作用和parallelize一样\n",
    "val rdd = sc.makeRDD(1 to 10,2)\n",
    "rdd.collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 二，常用Action操作\n",
    "\n",
    "Action操作将触发基于RDD依赖关系的计算。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**collect**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-15T04:01:02.232927Z",
     "start_time": "2019-09-15T04:00:54.486Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[0] at parallelize at <console>:27\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[0] at parallelize at <console>:27"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 10,5) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "all_data = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//collect操作将数据汇集到Driver,数据过大时有超内存风险\n",
    "val all_data = rdd.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**take**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[5] at parallelize at <console>:30\n",
       "part_data = Array(1, 2, 3, 4)\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(1, 2, 3, 4)"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//take操作将前若干个数据汇集到Driver，相比collect安全\n",
    "val rdd = sc.parallelize(1 to 10,5) \n",
    "val part_data = rdd.take(4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**takeSample**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[9] at parallelize at <console>:30\n",
       "sample_data = Array(5, 9, 10, 7, 4, 6, 3, 2, 8, 1)\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(5, 9, 10, 7, 4, 6, 3, 2, 8, 1)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//takeSample可以随机取若干个到Driver,第一个参数设置是否放回抽样\n",
    "val rdd = sc.parallelize(1 to 10,5) \n",
    "val sample_data = rdd.takeSample(false,10,0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**first**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[10] at parallelize at <console>:30\n",
       "first_data = 1\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//first取第一个数据\n",
    "val rdd = sc.parallelize(1 to 10,5) \n",
    "val first_data = rdd.first"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**count**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[11] at parallelize at <console>:30\n",
       "data_count = 10\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//count查看RDD元素数量\n",
    "val rdd = sc.parallelize(1 to 10,5)\n",
    "val data_count = rdd.count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**reduce**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[12] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "55"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//reduce利用二元函数对数据进行规约\n",
    "val rdd = sc.parallelize(1 to 10,5) \n",
    "rdd.reduce(_+_)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**foreach**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "55"
     ]
    },
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[24] at parallelize at <console>:33\n",
       "accum = LongAccumulator(id: 777, name: Some(sumAccum), value: 55)\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "LongAccumulator(id: 777, name: Some(sumAccum), value: 55)"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//foreach对每一个元素执行某种操作，不生成新的RDD\n",
    "//累加器用法详见共享变量\n",
    "val rdd = sc.parallelize(1 to 10,5) \n",
    "var accum = sc.longAccumulator(\"sumAccum\")\n",
    "rdd.foreach(x => accum.add(x))\n",
    "print(accum.value)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**countByKey**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pairRdd = ParallelCollectionRDD[25] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Map(1 -> 2, 2 -> 1, 3 -> 1)"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//countByKey对Pair RDD按key统计数量\n",
    "val pairRdd = sc.parallelize(Array((1,1),(1,4),(3,9),(2,16))) \n",
    "pairRdd.countByKey"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**saveAsTextFile和saveAsObjectFile**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "file = file:///Users/liangyun/CodeFiles/spark_tutorial/rdddata\n",
       "rdd = ParallelCollectionRDD[42] at parallelize at <console>:32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[42] at parallelize at <console>:32"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//saveAsTextFile保存rdd成text文件到本地,类似作用的\n",
    "val file = \"file:///Users/liangyun/CodeFiles/spark_tutorial/rdddata\"\n",
    "val rdd = sc.parallelize(1 to 5)\n",
    "rdd.saveAsTextFile(file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "2\n",
      "3\n",
      "4\n",
      "5\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "data = file:///Users/liangyun/CodeFiles/spark_tutorial/rdddata MapPartitionsRDD[45] at textFile at <console>:32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "file:///Users/liangyun/CodeFiles/spark_tutorial/rdddata MapPartitionsRDD[45] at textFile at <console>:32"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//重新读入\n",
    "val data = sc.textFile(file)\n",
    "data.collect.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "//saveAsObjectFile保存rdd成Object文件，类似的还有saveAsSequenceFile\n",
    "rdd3 = rdd.saveAsObjectFile(\"file:///Users/liangyun/CodeFiles/spark_tutorial/rdddata3\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 三，常用Transformation操作"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Transformation转换操作具有懒惰执行的特性，它只指定新的RDD和其父RDD的依赖关系，只有当Action操作触发到该依赖的时候，它才被计算。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**map**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[46] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//map操作对每个元素进行一个映射转换\n",
    "val rdd = sc.parallelize(1 to 10,3)\n",
    "rdd.collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(2, 3, 4, 5, 6, 7, 8, 9, 10, 11)"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.map(_+1).collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**filter**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[57] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(6, 7, 8, 9, 10)"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//filter应用过滤条件过滤掉一些数据\n",
    "val rdd = sc.parallelize(1 to 10,3)\n",
    "rdd.filter(_>5).collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**flatMap**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[50] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(Array(hello, world), Array(hello, China))"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//flatMap操作执行将每个元素生成一个Array后压平\n",
    "val rdd = sc.parallelize(Array(\"hello world\",\"hello China\"))\n",
    "rdd.map(_.split(\" \")).collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(hello, world, hello, China)"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.flatMap(_.split(\" \")).collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**sample**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[59] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(1, 2, 4, 8, 9, 10)"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//sample对原rdd在每个分区按照比例进行抽样,第一个参数设置是否可以重复抽样\n",
    "val rdd = sc.parallelize(1 to 10,1)\n",
    "rdd.sample(false,0.5,0).collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**distinct**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[61] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(4, 1, 5, 2, 3)"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//distinct去重\n",
    "val rdd = sc.parallelize(Array(1,1,2,2,3,3,4,5))\n",
    "rdd.distinct.collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**subtract**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[65] at parallelize at <console>:28\n",
       "b = ParallelCollectionRDD[66] at parallelize at <console>:29\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(4, 1, 2, 3)"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//subtract找到属于前一个rdd而不属于后一个rdd的元素\n",
    "val a = sc.parallelize(1 to 10)\n",
    "val b = sc.parallelize(5 to 15)\n",
    "a.subtract(b).collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**union** "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[71] at parallelize at <console>:31\n",
       "b = ParallelCollectionRDD[72] at parallelize at <console>:32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(1, 2, 3, 4, 5, 3, 4, 5, 6, 7, 8)"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//union合并数据\n",
    "val a = sc.parallelize(1 to 5)\n",
    "val b = sc.parallelize(3 to 8)\n",
    "a.union(b).collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**intersection**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[80] at parallelize at <console>:31\n",
       "b = ParallelCollectionRDD[81] at parallelize at <console>:32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(4, 5, 3)"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//intersection求交集\n",
    "val a = sc.parallelize(1 to 5)\n",
    "val b = sc.parallelize(3 to 8)\n",
    "a.intersection(b).collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**cartesian**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "boys = ParallelCollectionRDD[88] at parallelize at <console>:28\n",
       "girls = ParallelCollectionRDD[89] at parallelize at <console>:29\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((LiLei,HanMeiMei), (LiLei,Lily), (Tom,HanMeiMei), (Tom,Lily))"
      ]
     },
     "execution_count": 56,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//cartesian笛卡尔积\n",
    "val boys = sc.parallelize(Array(\"LiLei\",\"Tom\"))\n",
    "val girls = sc.parallelize(Array(\"HanMeiMei\",\"Lily\"))\n",
    "boys.cartesian(girls).collect\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**pipe**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[103] at parallelize at <console>:28\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[103] at parallelize at <console>:28"
      ]
     },
     "execution_count": 59,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//pipe可以调用外部可执行程序进行转换操作生成新的rdd，类似于hadoop streaming\n",
    "val rdd = sc.parallelize(Array(\"hello\",\"world\",\"hi\",\"apache\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "//getlen.py\n",
    "/*\n",
    "import sys\n",
    "for line in sys.stdin:\n",
    "    print(str(len(line.rstrip)))\n",
    "\n",
    "*/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd2 = PipedRDD[104] at pipe at <console>:27\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(5, 5, 2, 6)"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd2 = rdd.pipe(\"python getlen.py\")\n",
    "rdd2.collect"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**sortBy**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[97] at parallelize at <console>:31\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((4,1,1), (3,2,2), (1,2,3))"
      ]
     },
     "execution_count": 58,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//按照某种方式进行排序\n",
    "//指定按照第3个元素大小进行排序\n",
    "val rdd = sc.parallelize(Array((1,2,3),(3,2,2),(4,1,1)))\n",
    "rdd.sortBy(_._3).collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 四，常用PairRDD的转换操作"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "PairRDD指的是数据为Tuple2数据类型的RDD,其每个数据的第一个元素被当做key，第二个元素被当做value. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**reduceByKey**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[105] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((hello,4), (world,7))"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//reduceByKey对相同的key对应的values应用二元归并操作\n",
    "val rdd = sc.parallelize(Array((\"hello\",1),(\"world\",2),\n",
    "                               (\"hello\",3),(\"world\",5)))\n",
    "rdd.reduceByKey(_+_).collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**groupByKey**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[107] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((hello,CompactBuffer(1, 3)), (world,CompactBuffer(2, 5)))"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//groupByKey将相同的key对应的values收集成一个Iterator\n",
    "val rdd = sc.parallelize(Array((\"hello\",1),(\"world\",2),\n",
    "                               (\"hello\",3),(\"world\",5)))\n",
    "rdd.groupByKey().collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**sortByKey**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[117] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((world,2), (hello,1), (China,3), (Beijing,5))"
      ]
     },
     "execution_count": 66,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//sortByKey按照key排序,可以指定是否降序\n",
    "val rdd = sc.parallelize(Array((\"hello\",1),(\"world\",2),\n",
    "                               (\"China\",3),(\"Beijing\",5)))\n",
    "rdd.sortByKey(false).collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**join**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "age = ParallelCollectionRDD[136] at parallelize at <console>:31\n",
       "gender = ParallelCollectionRDD[137] at parallelize at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((HanMeiMei,(16,female)), (LiLei,(18,male)))"
      ]
     },
     "execution_count": 70,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//join相当于根据key进行内连接\n",
    "val age = sc.parallelize(Array((\"LiLei\",18),\n",
    "                        (\"HanMeiMei\",16),(\"Jim\",20)))\n",
    "val gender = sc.parallelize(Array((\"LiLei\",\"male\"),\n",
    "                        (\"HanMeiMei\",\"female\"),(\"Lucy\",\"female\")))\n",
    "age.join(gender).collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**leftOuterJoin和rightOuterJoin**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "age = ParallelCollectionRDD[141] at parallelize at <console>:32\n",
       "gender = ParallelCollectionRDD[142] at parallelize at <console>:34\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((HanMeiMei,(16,Some(female))), (LiLei,(18,Some(male))))"
      ]
     },
     "execution_count": 71,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//leftOuterJoin相当于关系表的左连接\n",
    "\n",
    "val age = sc.parallelize(Array((\"LiLei\",18),\n",
    "                        (\"HanMeiMei\",16)))\n",
    "val gender = sc.parallelize(Array((\"LiLei\",\"male\"),\n",
    "                        (\"HanMeiMei\",\"female\"),(\"Lucy\",\"female\")))\n",
    "age.leftOuterJoin(gender).collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "age = ParallelCollectionRDD[151] at parallelize at <console>:31\n",
       "gender = ParallelCollectionRDD[152] at parallelize at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((HanMeiMei,(Some(16),female)), (LiLei,(Some(18),male)))"
      ]
     },
     "execution_count": 73,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//rightOuterJoin相当于关系表的右连接\n",
    "val age = sc.parallelize(Array((\"LiLei\",18),\n",
    "                        (\"HanMeiMei\",16),(\"Jim\",20)))\n",
    "val gender = sc.parallelize(Array((\"LiLei\",\"male\"),\n",
    "                        (\"HanMeiMei\",\"female\")))\n",
    "age.rightOuterJoin(gender).collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**cogroup**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(a,(CompactBuffer(1, 3),CompactBuffer(2)))\n",
      "(b,(CompactBuffer(2),CompactBuffer(3, 5)))\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "x = ParallelCollectionRDD[160] at parallelize at <console>:32\n",
       "y = ParallelCollectionRDD[161] at parallelize at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[161] at parallelize at <console>:33"
      ]
     },
     "execution_count": 75,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//cogroup相当于对两个输入分别goupByKey然后再对结果进行groupByKey\n",
    "\n",
    "val x = sc.parallelize(Array((\"a\",1),(\"b\",2),(\"a\",3)))\n",
    "val y = sc.parallelize(Array((\"a\",2),(\"b\",3),(\"b\",5)))\n",
    "\n",
    "x.cogroup(y).collect.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**subtractByKey**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "x = ParallelCollectionRDD[164] at parallelize at <console>:32\n",
       "y = ParallelCollectionRDD[165] at parallelize at <console>:33\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((c,3))"
      ]
     },
     "execution_count": 76,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//subtractByKey去除x中那些key也在y中的元素\n",
    "\n",
    "val x = sc.parallelize(Array((\"a\",1),(\"b\",2),(\"c\",3)))\n",
    "val y = sc.parallelize(Array((\"a\",2),(\"b\",(1,2))))\n",
    "\n",
    "x.subtractByKey(y).collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**foldByKey**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "x = ParallelCollectionRDD[167] at parallelize at <console>:30\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((a,3), (b,10))"
      ]
     },
     "execution_count": 77,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//foldByKey的操作和reduceByKey类似，但是要提供一个初始值\n",
    "val x = sc.parallelize(Array((\"a\",1),(\"b\",2),(\"a\",3),(\"b\",5)),1)\n",
    "x.foldByKey(1)(_*_).collect\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 五，持久化操作"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "如果一个rdd被多个任务用作中间量，那么对其进行cache缓存到内存中对加快计算会非常有帮助。\n",
    "\n",
    "声明对一个rdd进行cache后，该rdd不会被立即缓存，而是等到它第一次被计算出来时才进行缓存。\n",
    "\n",
    "可以使用persist明确指定存储级别，常用的存储级别是MEMORY_ONLY和EMORY_AND_DISK。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "//cache缓存到内存中，使用存储级别 MEMORY_ONLY。\n",
    "//MEMORY_ONLY意味着如果内存存储不下，放弃存储其余部分，需要时重新计算。\n",
    "val a = sc.parallelize(1 to 10000,5)\n",
    "a.cache\n",
    "val sum_a = a.reduce((a,b)=>a+b)\n",
    "val cnt_a = a.count\n",
    "val mean_a = sum_a/cnt_a"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[186] at parallelize at <console>:39\n",
       "sum_a = 50005000\n",
       "cnt_a = 10000\n",
       "mean_a = 5000\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[186] at parallelize at <console>:39"
      ]
     },
     "execution_count": 96,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//persist缓存到内存或磁盘中，默认使用存储级别MEMORY_AND_DISK\n",
    "//MEMORY_AND_DISK意味着如果内存存储不下，其余部分存储到磁盘中。\n",
    "//persist可以指定其它存储级别，cache相当于persist(MEMORY_ONLY)\n",
    "import org.apache.spark.storage.StorageLevel.{MEMORY_AND_DISK}\n",
    "val a = sc.parallelize(1 to 10000,5)\n",
    "a.persist(MEMORY_AND_DISK)\n",
    "val sum_a = a.reduce(_+_)\n",
    "val cnt_a = a.count\n",
    "val mean_a = sum_a/cnt_a\n",
    "a.unpersist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 六，共享变量"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当spark集群在许多节点上运行一个函数时，默认情况下会把这个函数涉及到的对象在每个节点生成一个副本。\n",
    "\n",
    "但是，有时候需要在不同节点或者节点和Driver之间共享变量。\n",
    "\n",
    "Spark提供两种类型的共享变量，广播变量和累加器。\n",
    "\n",
    "广播变量是不可变变量，实现在不同节点不同任务之间共享数据。\n",
    "\n",
    "广播变量在每个机器上缓存一个只读的变量，而不是为每个task生成一个副本，可以减少数据的传输。\n",
    "\n",
    "累加器主要是不同节点和Driver之间共享变量，只能实现计数或者累加功能。\n",
    "\n",
    "累加器的值只有在Driver上是可读的，在节点上不可见。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "broads = Broadcast(157)\n",
       "rdd = ParallelCollectionRDD[187] at parallelize at <console>:38\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "100"
      ]
     },
     "execution_count": 97,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//广播变量 broadcast 不可变，在所有节点可读\n",
    "\n",
    "val broads = sc.broadcast(100)\n",
    "\n",
    "val rdd = sc.parallelize(1 to 10)\n",
    "rdd.map(_+broads.value).collect\n",
    "\n",
    "broads.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "total = LongAccumulator(id: 3803, name: Some(total), value: 55)\n",
       "rdd = ParallelCollectionRDD[189] at parallelize at <console>:39\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "55"
      ]
     },
     "execution_count": 98,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//累加器 只能在Driver上可读，在其它节点只能进行累加\n",
    "\n",
    "var total = sc.longAccumulator(\"total\")\n",
    "val rdd = sc.parallelize(1 to 10,3)\n",
    "\n",
    "rdd.foreach(x => total.add(x))\n",
    "total.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[190] at parallelize at <console>:37\n",
       "total = DoubleAccumulator(id: 3829, name: Some(sum), value: 505.0)\n",
       "count = LongAccumulator(id: 3830, name: Some(count), value: 100)\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "5.05"
      ]
     },
     "execution_count": 99,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(for(i<-1 to 100) yield 0.1*i )\n",
    "val total = sc.doubleAccumulator(\"sum\")\n",
    "var count = sc.longAccumulator(\"count\")\n",
    "\n",
    "rdd.foreach(x => {total.add(x)\n",
    "                  count.add(1)})\n",
    "total.value/count.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 七，分区操作"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "分区操作包括改变分区操作，以及针对分区执行的一些转换操作。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "coalesce：shuffle可选，默认为false情况下窄依赖，不能增加分区。repartition和partitionBy调用它实现。\n",
    "repartition：按随机数进行shuffle，相同key不一定在同一个分区\n",
    "partitionBy：按key进行shuffle，相同key放入同一个分区\n",
    "HashPartitioner：默认分区器，根据key的hash值进行分区，相同的key进入同一分区，效率较高，key不可为Array.\n",
    "RangePartitioner：只在排序相关函数中使用，除相同的key进入同一分区，相邻的key也会进入同一分区，key必须可排序。\n",
    "TaskContext:  获取当前分区id方法 TaskContext.get.partitionId\n",
    "mapPartitions：每次处理分区内的一批数据，适合需要分批处理数据的情况，比如将数据插入某个表，每批数据只需要开启一次数据库连接，大大减少了连接开支\n",
    "mapPartitionsWithIndex：类似mapPartitions，提供了分区索引，输入参数为（i，Iterator）\n",
    "foreachPartition：类似foreach，但每次提供一个Partition的一批数据\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**coalesce**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[193] at parallelize at <console>:37\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(Array(1, 2, 3), Array(4, 5, 6), Array(7, 8, 9, 10))"
      ]
     },
     "execution_count": 101,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//coalesce 默认shuffle为False，不能增加分区，只能减少分区\n",
    "//如果要增加分区，要设置shuffle = true\n",
    "//parallelize等许多操作可以指定分区数\n",
    "val a = sc.parallelize(1 to 10,3)  \n",
    "println(a.getNumPartitions)\n",
    "a.mapPartitions(iter => Iterator(iter.toArray)).collect\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "b = CoalescedRDD[195] at coalesce at <console>:34\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(Array(1, 2, 3), Array(4, 5, 6, 7, 8, 9, 10))"
      ]
     },
     "execution_count": 102,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val b = a.coalesce(2) \n",
    "b.mapPartitions(iter => Iterator(iter.toArray)).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**repartition**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[203] at parallelize at <console>:37\n",
       "c = MapPartitionsRDD[207] at repartition at <console>:38\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(Array(2, 7), Array(3, 4, 8), Array(5, 9), Array(1, 6, 10))"
      ]
     },
     "execution_count": 104,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//repartition按随机数进行shuffle，相同key不一定在一个分区，可以增加分区\n",
    "//repartition实际上调用coalesce实现，设置了shuffle = true\n",
    "val a = sc.parallelize(1 to 10,3)  \n",
    "val c = a.repartition(4) \n",
    "c.mapPartitions(iter => Iterator(iter.toArray)).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[209] at parallelize at <console>:36\n",
       "c = MapPartitionsRDD[213] at repartition at <console>:37\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(Array((a,1), (a,2), (c,3)), Array((a,1)))"
      ]
     },
     "execution_count": 105,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//repartition按随机数进行shuffle，相同key不一定在一个分区\n",
    "val a = sc.parallelize(Array((\"a\",1),(\"a\",1),(\"a\",2),(\"c\",3)))  \n",
    "val c = a.repartition(2)\n",
    "c.mapPartitions(iter => Iterator(iter.toArray)).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**partitionBy**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[218] at parallelize at <console>:44\n",
       "c = ShuffledRDD[219] at partitionBy at <console>:45\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(Array((b,2)), Array((a,1), (a,1), (c,3)))"
      ]
     },
     "execution_count": 107,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//partitionBy可以指定分区函数为 HashPartitioner或者RangePartitioner.\n",
    "//HashPartitioner为最常用分区器，根据key的hash值进行分区.\n",
    "//HashPartitioner相同的key进入同一分区，效率较高，key不可为Array.\n",
    "//RangePartitioner通常在排序相关场景应用。\n",
    "//RangePartitioner除相同的key进入同一分区，相邻的key也会进入同一分区。\n",
    "//RangePartitioner要求key必须可排序。\n",
    "\n",
    "import org.apache.spark.{HashPartitioner,RangePartitioner}\n",
    "val a = sc.parallelize(Array((\"a\",1),(\"a\",1),(\"b\",2),(\"c\",3)))  \n",
    "val c = a.partitionBy(new HashPartitioner(2))\n",
    "c.mapPartitions(iter => Iterator(iter.toArray)).collect()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**mapPartitions**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[221] at parallelize at <console>:41\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(Array(1, 2, 3, 4, 5), Array(6, 7, 8, 9, 10))"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//mapPartitions可以对每个分区分别执行操作\n",
    "//每次处理分区内的一批数据，适合需要按批处理数据的情况\n",
    "//例如将数据写入数据库时，可以极大的减少连接次数。\n",
    "//mapPartitions的输入分区内数据组成的Iterator，其输出也需要是一个Iterator\n",
    "//以下例子查看每个分区内的数据\n",
    "val a = sc.parallelize(1 to 10,2)\n",
    "a.mapPartitions(iter => Iterator(iter.toArray)).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**mapPartitionsWithIndex**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[223] at parallelize at <console>:40\n",
       "b = MapPartitionsRDD[224] at mapPartitionsWithIndex at <console>:41\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array(0|15, 1|40)"
      ]
     },
     "execution_count": 109,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//mapPartitionsWithIndex可以获取两个参数\n",
    "//即分区id和每个分区内的数据组成的Iterator\n",
    "val a = sc.parallelize(1 to 10,2)\n",
    "var b = a.mapPartitionsWithIndex{\n",
    "        (pid,iter) => {\n",
    "          var result = List[String]()\n",
    "            var i = 0\n",
    "            while(iter.hasNext){\n",
    "              i += iter.next()\n",
    "            }\n",
    "            result.::(pid + \"|\" + i).iterator\n",
    "           \n",
    "        }\n",
    "      }\n",
    "b.collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[227] at parallelize at <console>:41\n",
       "c = MapPartitionsRDD[228] at map at <console>:42\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((0,1), (1,2), (1,3), (2,4), (2,5))"
      ]
     },
     "execution_count": 111,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//利用TaskContext可以获取当前每个元素的分区\n",
    "import org.apache.spark.TaskContext\n",
    "\n",
    "val a = sc.parallelize(1 to 5,3)\n",
    "val c = a.map((TaskContext.get.partitionId,_))\n",
    "c.collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**foreachPartitions**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "total = DoubleAccumulator(id: 4281, name: Some(total), value: 199.0)\n",
       "a = ParallelCollectionRDD[229] at parallelize at <console>:48\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "199.0"
      ]
     },
     "execution_count": 112,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//foreachPartition对每个分区分别执行操作\n",
    "//范例：求每个分区内最大值的和\n",
    "var total = sc.doubleAccumulator(\"total\")\n",
    "\n",
    "val a = sc.parallelize(1 to 100,3)\n",
    "\n",
    "a.foreachPartition(iter =>{\n",
    "    val m = iter.reduce((x,y) => if(x>y) x else y)\n",
    "    total.add(m)\n",
    "})\n",
    "\n",
    "total.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**aggregate** "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[230] at parallelize at <console>:44\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "(210,20)"
      ]
     },
     "execution_count": 114,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//aggregate是一个Action操作\n",
    "//aggregate比较复杂，先对每个分区执行一个函数，再对每个分区结果执行一个合并函数。\n",
    "//例子：求元素之和以及元素个数\n",
    "//三个参数，第一个参数为初始值，第二个为分区执行函数，第三个为结果合并执行函数。\n",
    "val rdd = sc.parallelize(1 to 20,3)\n",
    "rdd.aggregate((0,0))((t,x)=>{println(t)\n",
    "                             (t._1+x,t._2+1)},\n",
    "                     (p,q)=>{println(p,q)\n",
    "                         (p._1+q._1,p._2+q._2)})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**aggregateByKey**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "a = ParallelCollectionRDD[231] at parallelize at <console>:43\n",
       "b = ShuffledRDD[232] at aggregateByKey at <console>:46\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array((c,2), (a,2), (b,3))"
      ]
     },
     "execution_count": 115,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//aggregateByKey的操作和aggregate类似，但是会对每个key分别进行操作\n",
    "//第一个参数为初始值，第二个参数为分区内归并函数，第三个参数为分区间归并函数\n",
    "\n",
    "val a = sc.parallelize(Array((\"a\",1),(\"b\",1),(\"c\",2),\n",
    "                             (\"a\",2),(\"b\",3)),3)\n",
    "\n",
    "val b = a.aggregateByKey(0)((a,b)=>if(a>b) a else b,\n",
    "                            (a,b)=>if(a>b) a else b)\n",
    "\n",
    "b.collect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Spark - Scala",
   "language": "scala",
   "name": "spark_scala"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "2.11.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}