开始在IDEA中编写代码（可以用spark实现原始的mapreduce
Spark on windows local
  异常信息：
    1. 17/05/20 09:32:08 ERROR SparkContext: Error initializing SparkContext.
        org.apache.spark.SparkException: A master URL must be set in your configuration ==> 
        设置一个master运行位置信息
    2. 17/05/20 09:33:22 INFO SparkContext: Successfully stopped SparkContext
        Exception in thread "main" org.apache.spark.SparkException: An application name must be set 
        in your configuration ==> 给定一个应用的名称
    3. null/bin/winutil.exe： windows环境没有配置hadoop的缘故的导致的
        ====> 只需要给定一个HADOOP_HOME的环境变量
    4. 可能出现源码方面的异常,一般情况提示为NullPointException,解决方案：修改hadoop底层源码 --> 
    5、可以看到SparkUI界面：4040
    6、甚至可以把路径给定本地的路径data/XXX.txt
        输出路径 result/wc/....
        直接连HDFS都不用开启
==================================================
Spark on yarn
    1、记得把setMaster("local")注释掉
    2、打包
        1、maven打包,在view中ToolWindow里面找到Maven打开
        2、file -> project Structure -> Artifacts -> + -> 
        jar -> from modules DeXXX -> OK(什么都不用选,直接OK) -> 
        把相关依赖包全删了,留一个out -》修改下Output directory的路径为E:\MySalcaWorkSpace\spark
        (这里是你的项目名)\out\ 
        选择ok -> 然后在上面工具栏找到Build -> Build Artifacts -> build(编译)或者rebuild(重新编译)
            http://spark.apache.org/docs/1.6.1/running-on-yarn.html
===========yarn  模式的相关配置==================
    1、yarn的配置信息(yarn-site.xml)在spark的classpath中,这个其实我们已经配置过了
    （就是hadoop的etc的目录
    官网关于submit的解释：http://spark.apache.org/docs/1.6.1/submitting-applications.html
    local执行：(如果我们不给定master的值,默认是本地)
    bin/spark-submit \
    --class com.ibeifeng.bigdata.spark.app.core.SparkWordCount \     //idea里 右键copy reference
    /home/beifeng/logs-analyzer.jar 
    
===================standalone模式=============================
    bin/spark-submit \
    --master spark：//bigdata-03:6066
    --deploy-mode cluster \
    --class com.ibeifeng.bigdata.spark.app.core.SparkWordCount \
    /home/beifeng/logs-analyzer.jar 

==================yarn模式提交====================================
    bin/spark-submit \
    --master yarn \
    --deploy-mode client \     driver 为指定服务器
    --class com.ibeifeng.bigdata.spark.app.core.SparkWordCount \
    /home/beifeng/logs-analyzer.jar 

    bin/spark-submit \
    --master yarn \  
    --deploy-mode cluster \     driver 为cluster
    --class com.ibeifeng.bigdata.spark.app.core.SparkWordCount \
    /home/beifeng/logs-analyzer.jar 

    
    注意区别：1、这里如果是client会打印具体日志信息,如果是cluster就不打印,就打印等待信息
              2、如果是client运行,driver会在本机开启（提交任务的那台机器,如果是cluster运行,
              driver就交给RM（Master去调度,找有资源的机器去开启
          
====如果是cluster,推荐使用REST API提交 rest api 提交应用 ====

spark2.1 RSET API参考：http://spark.apache.org/docs/2.2.0/monitoring.html#rest-api
    
    curl -X POST http://make.spark.com:6066/v1/submissions/create \
    --header "Content-Type:application/json;charset=UTF-8" --data '{
  "action" : "CreateSubmissionRequest",
  "appArgs" : [ "args1, args2,..." ], 
  "appResource" : "file:/myfilepath/spark-job-1.0.jar", 
  "clientSparkVersion" : "2.1.0",
  "environmentVariables" : {
    "SPARK_ENV_LOADED" : "1"
  },
  "mainClass" : "com.mycompany.MyJob",
  "sparkProperties" : {
    "spark.jars" : "file:/myfilepath/spark-job-1.0.jar",
    "spark.driver.supervise" : "false",
    "spark.app.name" : "MyJob",
    "spark.eventLog.enabled": "true",
    "spark.submit.deployMode" : "cluster",
    "spark.master" : "spark://spark-cluster-ip:6066"
  }
}'

        
spark-cluster-ip ==>spark master地址  默认端口为6060,如果被占用会依次查找6067,6068
action ==>执行的动作为 CreateSubmissionRequest创建一个任务,提交任务
appArgs  ==>传入的参数列表
appResource ==> jar包的位置   xxx.jar
clientSparkVersion  ==>spark版本
environmentVariables  ==>是否加载本地环境变量
mainClass : com.mycompany.MyJob ==> 程序的主类
sparkProperties : {…}  ==> spark的参数配置

Spark应用的构成：
    master + worker
    
    master：基于standalone的Spark集群,Cluster Manger就是Master,Master负责分配资源,在集群启动时,
    Driver向Master申请资源
    worker:负责监控自己节点的内存和CPU,并向master汇报,Worker默认情况下分配一个Executor,配置时根据需
    要也可以配置多个Executor,worker保存了Executor的句柄,根据需要可以kill掉Executor进程
    
    Driver + Executors
    
    Driver: main方法的运行的jvm的地方；主要功能是：SparkContext上下文创建、RDD构建、RDD调度、RDD运行
    资源调度,Driver由框架直接生成。
    Executor：具体task执行的jvm的地方,RDD的API就是在这里运行的,这里执行的才是真正的业务逻辑代码

  
=====driver、executors、master、worker====== 
 1．驱动器节点(Driver)
Spark的驱动器是执行开发程序中的 main方法的进程,用来创建SparkContext、创建 RDD,
以及进行 RDD 的转化操作和行动操作代码的执行。对于spark shell,当启动 Spark shell的时候,系统会自启一
个Spark驱动器程序,也就是事先创建加载的一个sc的 SparkContext 对象。驱动器程序终止,Spark 应用就结束了

Driver在spark作业执行时主要负责以下操作：
1)把用户程序转为任务
    Driver程序负责把用户程序转为多个物理执行的单元,单元也就是任务(task),从上层来看,spark程序的流程：
--1、读取或者转化数据创建一系列 RDD,然后使用转化操作生成新的RDD,最后使用行动操作得到结果或者将数据
存储到文件存储系统中
--2、Spark程序会隐式地创建一个由上述操作组成的逻辑上的DAG图(有向无环图)。当Driver序运行时,它会把这个
逻辑图转为物理执行计划。
--3、Spark 会对逻辑执行计划作一些优化,比如连续的映射转为流水线化执行,将多个操作合并到一个步骤中等。
这样Spark就把逻辑计划转为一系列步骤(stage),而每个stage又由多个task组成。这些task会被打包并送到集群中
,而task是Spark中最小的任务执行单元,用户程序通常要启动成百上千的独立任务。

2)跟踪Executor的运行状况
    有了物理执行计划之后,Driver程序必须在各个Executor进程间协调任务的调度。Executor进程启动后,会向
    Driver进程注册自己。因此,Driver进程就可以是时时跟踪监控应用中所有的Executor节点的运行信息。

3)为执行器节点调度任务
    Driver程序会根据当前的Executor节点集合,尝试把基于数据所在位置分配task给合适的Executor进程,当Task
执行时,Executor进程会把缓存数据存储起来,而Driver进程也会跟踪这些缓存数据的位置,并且利用这些位置信息
来调度以后的任务,以尽量减少数据的网络IO。

4)UI展示应用运行状况
Driver程序会将一些 Spark 应用的运行时的信息通过网页界面呈现出来,默认为端口4040。比如,在本地模式中,访
问http://localhost:4040,就可以查看这个UI界面

2．执行器节点(Executor)

    Spark Executor节点是一个工作进程,负责在 Spark 作业中运行任务,任务间相互独立。Spark 应用启动时,
Executor节点被同时启动,并且始终伴随着整个 Spark 应用的生命周期而存在。如果有Executor节点发生了故障
或崩溃,Spark 应用也可以继续执行,会将出错节点上的任务调度到其他Executor节点上继续运行。

执行器进程有两大作用：
1、它们负责运行组成 Spark 应用的任务,并将结果返回给驱动器进程；
2、它们通过自身的块管理器（Block Manager为用户程序中要求缓存的 RDD 提供内存式存储。RDD 是直接缓存
在Executor进程内的,因此任务可以在运行时充分利用缓存数据加速运算。执行器程序通常都运行在专用的进程中。 
  
  

3.driver与executor内部的运行逻辑及调配

编程的Spark程序,打包提交到Driver端,这样就构成了一个Driver 
1、在Driver中,RDD首先交给DAGSchedule进行(步骤)Stage的划分。 
2、由底层的调度器TaskScheduler就与Executor进行交互,Driver和上图中3个Worker节点的Executor发指令,让它
们在各自的线程池中运行Job。 
3、运行时Driver能获得Executor的具体运行资源,这样Driver与Executor之间进行通信,通过网络的方式,Driver把
划分好的Task传送给Executor,Task就是我们的Spark程序的业务逻辑代码,所以上面说,是在executor中。
4、Executor接收任务,进行反序列化,得到数据的输入和输出,在分布式集群的相同数据分片上,数据的业务逻辑一
样,只是数据不一样罢了,然后由Executor的线程池负责执行具体的task

spark on yarn cluster

spark on yarn client

  
========Spark应用启动配置信息可以在三个地方配置=========
  1. spark-defaults.conf
  2. spark-submit脚本参数：如下
    --name 给定job的名称（可以在代码中给定,也可以在这里给定
    --master：给定运行spark应用的执行位置信息
    --conf：给定配置参数,可以有多个--conf
    --propertise-file：配置信息文件,默认在conf/spark-defaults.conf
  3. spark应用中通过SparkConf对象指定参数(代码)
  优先级：1 < 2 < 3
  
配置参数官网：http://spark.apache.org/docs/2.2.0/configuration.html#available-properties

spark-submit脚本参数 ==> Spark资源调优

--master：给定运行spark应用的执行位置信息(yarn,http://localhost:6066)
--deploy-mode：给定driver在哪儿执行(client、cluster)
  client：driver在执行spark-submit的那台机器上运行
  cluster：driver在集群中根据资源分配一台机器运行,
--driver-memory MEM：指定driver运行的时候jvm的内存大小,默认1G,一般情况下要求比单个executor的内存要大
--executor-memory MEM：指定单个executor的内存大小,默认1G
--driver-cores NUM: 指定spark on standalone的时候,而且是cluster模式的请看看下,driver运行过程中使用的
  core数量,默认为1
--supervise: 当运行环境为standalone/mesos + cluster,如果driver运行失败,会重新自动进行恢复操作,client
  模式就不会
--total-executor-cores NUM ：运行环境为standalone/mesos,给定应用需要的总的core的数目,默认所有
--executor-cores NUM：运行环境为standalon/yarn,给定应用运行过程中,每个executor包含的core数目,
  默认1个(yarn),默认all(standalone)
--driver-cores NUM：spark on yarn cluster, 给定driver运行需要多少个core,默认1个
--num-executors NUM: 申请多少个executor,默认2,其实这里的executor也就是yarn平台上container容器

====================Spark on yarn job history配置=====================
官网：http://spark.apache.org/docs/2.2.0/running-on-yarn.html

  -1. 配置在yarn页面可以通过链接直接点击进入history执行页面
    --1. 修改yarn-site.xml文件,然后重启yarn
  <property>
    <name>yarn.log.server.url</name>
    <value>http://make.spark.com:19888/jobhistory/job/</value>
  </property>      
  <property>
    <name>yarn.log-aggregation-enable</name>
    <value>true</value>
  </property>
     --2. 修改spark-defaults.conf
      spark.yarn.historyServer.address        http://make.spark.com:18080
      
  -2. 启动spark的history server
    sbin/start-history-server.sh
    http://hadoop-senior01:18080/

      
========用spark实现uv,pv的统计(使用page_views.data测试)=======

概念介绍：
DAG Scheduler 有向无环图的资源调度
YarnClusterScheduler yarn集群的资源调度
StandaloneScheduler standalone模式下的资源调度
ExecutorBackend是资源通信

Spark应用构建及提交流程：
  -1. Driver中RDD的构建
  -2. RDD Job被触发(需要将rdd的具体执行步骤提交到executor中执行)
  -3. Driver中的DAGScheduler将RDD划分为Stage阶段
  -4. Driver中的TaskScheduler将一个一个stage提交到executor上执行

Spark应用的执行过程
  -1. client向资源管理服务(ResourceManager、Master等)申请运行的资源(driver资源)
      注意：如果是client模式下,driver的资源不用进行申请操作
  -2. 启动driver
  -3. driver向资源管理服务(ResourceManager、Master等)申请运行的资源(executor资源)
  -4. 启动executor
  -5. rdd构建
  -6. rdd执行

===================RDD到底是什么====================== 
what is RDD？
Resilient Distributed Datasets=>弹性分布式数据集
默认情况下：每一个block对应一个分区,一个分区会开启一个task来处理
Resilient：可以存在给定不同数目的分区、数据缓存的时候可以缓存一部分数据也可以缓存全部数据
Distributed：分区可以分布到不同的executor执行(也就是不同的worker/NM上执行)
Datasets：内部存储是数据
RDD中的数据是不可变的、是分区的；

RDD的五大特性：见ppt
    
1、一组分片(a list Partition)
    即数据集的基本组成单位。对于RDD来说,每个分片都会被一个计算任务处理,并决定并行计算的粒度。用户
可以在创建RDD时指定RDD的分片个数,如果没有指定,那么就会采用默认值。默认值就是程序所分配到的CPU Core
的数目。

2、一个计算每个分区的函数(a function for computting each split)
    Spark中RDD的计算是以分片为单位的,每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器
    进行复合,不需要保存每次计算的结果。

3、RDD之间的依赖关系(a list of dependences on other RDDs)
    RDD的每次转换都会生成一个新的RDD,所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数
    据丢失时,Spark可以通过这个依赖关系重新计算丢失的分区数据,而不是对RDD的所有分区进行重新计算。

4、一个Partitioner,即RDD的分片函数(a parationer for key-value RDDs)
    当前Spark中实现了两种类型的分片函数,一个是基于哈希的HashPartitioner,另外一个是基于范围的
    RangePartitioner。只有对于于key-value的RDD,才会有Partitioner,非key-value的RDD的Parititioner的值
    是None。Partitioner函数不但决定了RDD本身的分片数量,也决定了parent RDD Shuffle输出时的分片数量。

5、一个列表,存储存取每个Partition的优先位置(preferred location(s))
    对于一个HDFS文件来说,这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”
    的理念,Spark在进行任务调度的时候,会尽可能地将计算任务分配到其所要处理数据块的存储位置。因为有副
    本的存在,所以有可能返回多个最佳位置

RDD的创建
  -1. 外部数据(非内存数据)：基于MapReduce的InputFormat进行创建
    sc.textFile ==> 底层使用TextInputFormat读取数据形成RDD；使用旧API
        源码：SparkContext类中828行,这里使用的hadoopFile是hadoop的旧API
            def textFile(
        path: String,
        minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
        assertNotStopped()
        hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
        minPartitions).map(pair => pair._2.toString).setName(path)
        }
    
    sc.newAPIHadoopFile ==> 底层使用TextInputFormat读取数据形成RDD；使用新API
        SparkContext类中828行,1125行,这里使用新API：
        def newAPIHadoopRDD[K, V, F <: NewInputFormat[K, V]](
        conf: Configuration = hadoopConfiguration,
        //fClass就是NewInputFormat,所以我也可以调用这个API手动给定输入的格式化器
        fClass: Class[F],
        kClass: Class[K],
        vClass: Class[V]): RDD[(K, V)] = withScope {
        assertNotStopped()
        // Add necessary security credentials to the JobConf. Required to access secure HDFS.
        val jconf = new JobConf(conf)
        SparkHadoopUtil.get.addCredentials(jconf)
        new NewHadoopRDD(this, fClass, kClass, vClass, jconf)
        }
        
  -2. 内存中数据：基于序列化进行创建,如下
    scala> val seq = List(1,2,3,4,5,6,7)
    seq: List[Int] = List(1, 2, 3, 4, 5, 6, 7)

    scala> val rdd2 = sc.parallelize(seq)
    rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:29
    
    
    把RDD源码打开,看几个函数：五大特性,指的就是这五个抽象的API
        116行： 计算分区的函数
                def compute(split: Partition, context: TaskContext): Iterator[T]
                
                //获取分区信息
                protected def getPartitions: Array[Partition]
                
                //获取依赖
                protected def getDependencies: Seq[Dependency[_]] = deps
                
                //获取最优的路径信息 
                protected def getPreferredLocations(split: Partition): Seq[String] = Nil
                
                //分区器
                @transient val partitioner: Option[Partitioner] = None
    你会发现textFile创建RDD的API要么是new HadoopRDD要么就是new NewHadoopRDD,可以详细看下这两个个源码
    compute方法

=============RDD构建底层原理========================

  -1. RDD分区数量 == InputFormat的getsplit方法返回的集合中split的数量(分区数量和块的数量是一样的)
  -2. RDD中不包含数据,只包含数据存储的位置信息,比如split  
    总结：
        （分区数量默认的情况下,rdd一个block块是一个分区,但是前提是文件足够大,如果文件很小,最小分区数
        量为2
    首先：RDD这个类被HadoopRDD实现,获取HDFS上的分片,数据等信息
          rdd的不储存数据的,只是能够指定的找到数据的位置
    
    疑问：那RDD相互依赖,下层的RDD的数据又是怎么来的呢
    解答：当一个RDD生成下一个RDD的时候通过MapPartitionsRDD这个类,也是调用compute方法去找父类传过来的
    迭代器
        我们可以在RDD这个类当中看到许多算子操作都会生成new MapPartitionsRDD等等类型的RDD
        并且下一个RDD也会记录着上一个RDD的信息
    
    疑问：那shuffle之后的RDD的数据应该已经被改变和移动位置了？
    解答：ShuffledRDD的compute方法里面获取了从某个分片到另一个分片的数据,
    
    疑问：那么这数据到底存在哪里？
    解答：之前的一顿操作之后,数据被读到了内存中,变成BlockRDD内存块RDD
          内存块RDD通过BlockManager获取在内存中块RDD的id,然后进行转换
          其实就是RDD当中只有各式各样的块信息,位置信息,当真正使用的时候才会去拿数据,在内存中处理
    
    疑问：为什么我的DAG图有的是灰色,有的是彩色的
    解答：灰色的证明之前有类似的操作,在内存中是有缓存信息的,所以没有重复执行
    

    spark.default.parallelism    (默认值就是2)
    For distributed shuffle operations like reduceByKey and join, the largest number of partitions 
    in a parent RDD. For operations like parallelize with no parent RDDs, it depends on the cluster 
    manager:
    
    Local mode: number of cores on the local machine
    Mesos fine grained mode: 8
    Others: total number of cores on all executor nodes or 2, whichever is larger
    
    本地模式：默认为本地机器的CPU数目,若设置了local[N],则默认为N
    Apache Mesos：默认的分区数为8
    Standalone或YARN：默认取集群中所有核心数目的总和,或者2,取二者的较大值
================分区结论============
    对于parallelize来说,没有在方法中的指定分区数,则默认为spark.default.parallelism
    对于textFile来说,没有在方法中的指定分区数,则默认为min(defaultParallelism,2),而defaultParallelism
    对应的就是spark.default.parallelism。如果是从hdfs上面读取文件,其分区数为文件分片数(128MB/片)
    
    eg.案例测试
        测试：当要做某些数据累加的时候,有多个分区的数据,会有什么影响？
        需求：想把所有的(key,value)里面的key较小的value都累加到key较大的上面去,如(3, 6),(5,4),(6, 2)
              结果(6,2+4+6=12),(5,4+6=10),(3,6)
              结果(6,2+4+6=12),(5,4+6=10),(3,6)
            val data = Array((3, 6),(5,4),(6, 2)) //key升序  自然顺序排序
        
        //设置2分区 这里不设置 则取机器所在的核心数和2的较大值,或为local[N],中的N与2的较大值为分区数
            val rdd1 = sc.parallelize(data, 2) 
            var sum = 0  
            rdd1.foreach(f => {  
            println((f._1,sum+f._2) + "--" + sum)  
            sum += f._2  
            })  
        结果：(3,6)--0  
              (5,4)--0  
              (6,6)--4 
        注意到(3,6)和(5,4)都是0,我们猜测它们的执行代码不在同台机器上(确切地说应该是不在同一分区),我
        们使用glom来验证下,glom函数会把RDD分区数据组装到数组类型的RDD中
        结果rdd1.glom.collect  
        res28: Array[Array[(Int, Int)]] = Array(Array((3,6)), Array((5,4), (6,2)))  
        Ok,我们的猜想得到了验证,果然它们不在一个分区中。问题暴力简单的解决方法是使用repartition把分
        区数置1
        (3,6)--0
        (5,10)--6
        (6,12)--10
====================================================    
RDD的方法类型(API类型)
    你可以分为2种或者分为3种
        2种：--lazy（tranformation和缓存,只有被调用的时候才会执行
             --立即执行的action,和清除缓存的操作
        3种：tranformation和action和persistent

        
    transformation(transformation算子)：转换操作
      功能：由一个RDD产生一个新的RDD,不会触发job的执行
      在这些类型的API调用过程中,只会构建RDD的依赖,也称为构建RDD的执行逻辑图(DAG图)
      这个操作是在driver过程中执行的,当有action的操作时,就会把对应的信息发送到excutor上面
    
    action(action算子):动作/操作
      功能：触发rdd的job执行提交操作,并将rdd对应的job提交到executor上执行
      该类型的API调用的时候,会触发job的执行,并将job的具体执行过程提交到executor上执行,最终的执行结果
      要不输出到其它文件系统或者返回给driver,所以这也是driver的内存大小要比单个的excutor内存大的原因
    
persist:(RDD缓存/RDD持久化)
    rdd将数据进行缓存操作或者清除缓存的rdd数据或者数据进行了checkpoint(只在streaming中使用)
    rdd.cache() 数据缓存到内存中
    rdd.persist(xxx) 数据缓存到指定级别的存储系统中(内存\内存+磁盘\磁盘)
    rdd.unpersist() 清除缓存数据
    
    ======缓存级别  四种常用的(结尾带_2的都会默认保存副本)=======

    1.MEMORY_ONLY
        使用未序列化的Java对象格式,将数据保存在内存中。如果内存不够存放所有的数据,则数据可能就不会
        进行持久化。那么下次对这个RDD执行算子操作时,那些没有被持久化的数据,需要从源头处重新计算一遍
        这是默认的持久化策略,使用cache()方法时,实际就是使用的这种持久化策略。

    2.MEMORY_AND_DISK
        使用未序列化的Java对象格式,优先尝试将数据保存在内存中,如果内存不够存放所有的数据,会将数据写
        入磁盘文件中,下次对这个RDD执行算子时,持久化在磁盘文件中的数据会被读取出来使用。

    3.MEMORY_ONLY_SER
        基本含义同MEMORY_ONLY,唯一的区别是,会将RDD中的数据进行序列化,RDD的每个partition会被序列化成
        一个字节数组。这种方式更加节省内存,从而可以避免持久化的数据占用过多内存导致频繁GC。
    4.MEMORY_AND_DISK_SER
        基本含义同MEMORY_AND_DISK,唯一的区别是,会将RDD中的数据进行序列化,RDD的每个partition会被序列
        化成一个字节数组。这种方式更加节省内存,从而可以避免持久化的数据占用过多内存导致频繁GC。
    
    ======缓存级别使用策略=======
    
    默认情况下,性能最高的当然是MEMORY_ONLY,但前提是你的内存必须足够足够大,可以绰绰有余地存放下整个
    RDD的所有数据。
    
    如果使用MEMORY_ONLY级别时发生了内存溢出,那么建议尝试使用MEMORY_ONLY_SER级别。
    该级别会将RDD数据序列化后再保存在内存中,此时每个partition仅仅是一个字节数组而已,大大减少了对象数
    量,并降低了内存占用。这种级别比MEMORY_ONLY多出来的性能开销,主要就是序列化与反序列化的开销。
    但是后续算子可以基于纯内存进行操作,因此性能总体还是比较高的。此外,可能发生的问题同上,如果RDD中的
    数据量过多的话,还是可能会导致OOM内存溢出的异常。

    如果纯内存的级别都无法使用,那么建议使用MEMORY_AND_DISK_SER策略,而不是MEMORY_AND_DISK策略。因为既
    然到了这一步,就说明RDD的数据量很大,内存无法完全放下。序列化后的数据比较少,可以节省内存和磁盘的空
    间开销。同时该策略会优先尽量尝试将数据缓存在内存中,内存缓存不下才会写入磁盘。
    
    
    =======缓存相关概念========
    缓存可以在DAG图中看到小绿点,证明不是从最原始的数据源拿的数据,也可以4040的Storage界面当中看到有缓
    存的rdd信息,可以看到缓存级别
    
    缓存是以分区为单位,就是说,并不是把这个rdd里所有数据缓存,他只会缓存分区里的数据（可以是多个分区
    如果你只是做first这样的操作,也不会完全缓存,因为你只输出第一个分区的头部,所以第二个分区没有缓存
    但是collect就会全部做缓存
    
    那么他到底缓存多少,还要看内存机制中,还剩多少空间,如果只存了百分之50,那另外百分之50也只能从数据源
    拿,并且当你cache的时候,Storage界面是什么都没有显示的(因为没存,只有当你对这个缓存进行操作的时候
    (这时候才存的,才有显示
    所以缓存是lazy操作,但是清除缓存就是立即执行的,数据如果缓存在内存中自然很快,如果是磁盘中的话,因为
    是缓存在本地磁盘中,所以速度也比从别的机器上更快