Spark RDD持久化

spark上下文      

package com.mao.scala

import org.apache.spark.{SparkConf, SparkContext}

object WordCountDemo {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf();
    conf.setAppName("WordCountSpark");
   // 本地后台调度器
    conf.setMaster("local");
   // conf.setMaster("local[3]");     //3线程,模拟cluster集群
   // conf.setMaster("local[*]");     //匹配cpu个数，
   // conf.setMaster("local[3,2]");   //3:3个线程，2最多重试次数

    //[相当于伪分布式]   
    //conf.setMaster("local-cluster[N, cores, memory]");  //模拟spark集群。 

    //[完全分布式]
    //StandaloneSchedulerBackend
    //conf.setMaster("spark://s201:7077");    //连接到spark集群上.

    val sc = new SparkContext(conf);
    val rdd1 = sc.textFile("d:/mr/test.txt");
    val rdd2 = rdd1.flatMap(line => line.split(" "));
    val rdd3 = rdd2.map(( _ ,1));
    val rdd4 = rdd3.reduceByKey(_ + _);
    val r = rdd4.collect();
    r.foreach(println);
  }
}
           

RDD持久化

    跨操作进行RDD的内存式存储。

    持久化RDD时，节点上的每个分区都会保存操内存中,以备在其他操作中进行重用。

    缓存技术是迭代式计算和交互式查询的重要工具。

    使用persist()和cache()进行rdd的持久化。

    cache()是persist()一种.

    action第一次计算时会发生persist().

    spark的cache是容错的，如果rdd的任何一个分区丢失了，都可以通过最初创建rdd的进行重新计算。

    persist可以使用不同的存储级别进行持久化。

    MEMORY_ONLY            //只在内存

    MEMORY_AND_DISK

    MEMORY_ONLY_SER        //内存存储(串行化)

    MEMORY_AND_DISK_SER 

    DISK_ONLY            //硬盘

    MEMORY_ONLY_2        //带有副本 

    MEMORY_AND_DISK_2    //快速容错。

    OFF_HEAP 

删除持久化数据: rdd.unpersist();

数据传递:

     map(),filter()高级函数中访问的对象被串行化到各个节点。每个节点都有一份拷贝。

    变量值并不会回传到driver程序。

共享变量

spark通过广播变量和累加器实现共享变量。

    [广播变量]

        //创建广播变量

        val bc1 = sc.broadcast(Array(1,2,3))

        bc1.value

    [累加器]

        val ac1 = sc.longaccumulator("ac1")

        ac1.value

        sc.parell..(1 to 10).map(_ * 2).map(e=>{ac1.add(1) ; e}).reduce(_+_)

        ac1.value            //10

    通过spark实现pi的分布式计算

    ----------------------------

    sc.parallelize(1 to 20).map(e=>{val a = 1f / (2 * e - 1) ;val b = if (e % 2 == 0) -1 else 1 ;a * b * 4}).reduce(_+_)