Spark程序设计——accumulator、广播变量、cache

    xiaoxiao2021-12-10  8

    (一)accumulator(累加器、计数器) 类似于MapReduce中的counter,将数据从一个节点发送到其他各个节点上去。 通常用于监控,调试,记录符合某类特征的数据数目等。 –分布式counter Accumulator使用

    import SparkContext._ val total_counter=sc.accumulator(OL,"total_counter")//第一个参数表示数字0和long,第二个参数是counter的名字 //定义两个累加器 val counter0=sc.accumulator(OL,"counter0") val counter1=sc.accumulator(OL,"counter1") val count=sc.parallelize(1 to n,slices).map{i=>total_counter+=1 val x=random*2-1 val y=random*2-1 if(x*x+y*y<1){counter1 +=1 }//累加器counter1加1 else{ counter0 +=1//累加器counter0加1 } if(x*x+y*y<1)1 else 0 }.reduce(_+_)

    (二)广播变量 广播机制 – 高效分发大对象,比如字典(map),集合(set)等,每个executor一份,而不是每个task一份。

    HttpBroadcast与TorrentBroadcast(基于ptop协议,例子电驴下载) 不使用广播–低效

    val data=Set(1,2,4,6,.....)//大小为128MB val rdd=sc.parallelize(1 to 6,2) val observedSizes=rdd.map(_=>data.size)

    使用广播高效

    val data=Set(1,2,4,6,....)//大小为128MB val bdata=sc.broadcast(data)//将大小为128MB的Set广播出去 val rdd=sc.parallelize(1 to 1000000,100) val observedSizes=rdd.map(_=>bdata.value.size.....) //在各个task中,通过bdata.value获取广播的集合

    (三)cache基本概念和使用

    Spark RDD Cache __允许将RDD缓存到内存中或磁盘上,以便重用。 __Spark提供了多种缓存级别,以便于用户根据实际需求进行调整。

    RDD cache的使用

    val data=sc.textFile("hdfs://nn:8020/input") data.cache() //data.persist(StorageLevel.DISK_ONLY_2)

    深入cache //下面两段代码的区别在于cache的使用:

    val data=sc.textFile("hdfs://nn:8020/input") data.cache() data.filter(_.startWith("error")).count data.filter(_.endWith("hadoop")).count data.filter(_.startWith("hbase")).count ......

    从磁盘上读取1次,之后从内存中读取(在遇到action时执行,将数据加载到内存中)

    val data=sc.textFile("hdfs://nn:8020/input") data.filter(_.startWith("error")).count data.filter(_.endWith("hadoop")).count data.filter(_.startWith("hbase")).count ......

    每次从磁盘上读取数据

    转载请注明原文地址: https://ju.6miu.com/read-700186.html

    最新回复(0)