spark rdd java_Spark之RDD（含Java运行环境配置）

一：RDD简介

(一)RDD概念

RDD(Resilient Distributed DataSet)，弹性分布式数据集，是Spark中最基本，也是最重要的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知度调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能重用工作集，这极大地提升了查询速度。因为有RDD，所以Spark才支持分布式的计算。RDD由分区组成。

(二)RDD的五个特性

(1)一组分片(Partition)，即数据集的基本组成单位。---RDD会被分片处理，用于并行计算

对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。

用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。

默认值就是程序所分配到的CPU Core的数目。

(2)一个计算每个分区的函数。---一个对每个split(数据分区)进行计算的函数，也称为RDD的算子

Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。

compute函数会对迭代器进行复合，不需要保存每次计算的结果。

(3)RDD之间的依赖关系。(DAG有向无环图调度构造依赖关系)

RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。

在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。

(4)一个Partitioner(分区器)，即RDD的分片函数。---用来对RDD的数据做手动分区

当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。

只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。

Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。

(5)一个列表，存储存取每个分片(Partition)的优先位置(preferred location)。

对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。

按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

(三)RDD原理

(四)wordcount程序RDD执行流程

二：RDD创建方式

(一)通过读取文件生成

由外部存储系统的数据集创建，包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等

val file = sc.textFile("/spark/input/c.txt")

(二)通过并行化的方式创建RDD

val array = Array(1,2,3,4,5)

val rdd= sc.parallelize(array)

(三)其他方式

读取数据库等等其他的操作。也可以生成RDD。

RDD可以通过其他的RDD转换而来的。

三：RDD编程API

Spark支持两个类型(算子)操作：Transformation和Action

(一)Transformation---不会触发计算，延时加载(计算)

主要做的是就是将一个已有的RDD生成另外一个RDD。Transformation具有lazy特性(延迟加载)。Transformation算子的代码不会真正被执行。只有当我们的程序里面遇到一个action算子的时候，代码才会真正的被执行。这种设计让Spark更加有效率地运行。

转换

含义

map(func)

返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成

filter(func)

返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成

flatMap(func)

类似于map，但是每一个输入元素可以被映射为0或多个输出元素(所以func应该返回一个序列，而不是单一元素)

mapPartitions(func)

类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]

mapPartitionsWithIndex(func)

类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是

(Int, Interator[T]) => Iterator[U]

sample(withReplacement, fraction, seed)

根据fraction指定的比例对数据进行采样，可以选择是否使用随机数进行替换，seed用于指定随机数生成器种子

union(otherDataset)

对源RDD和参数RDD求并集后返回一个新的RDD

intersection(otherDataset)

对源RDD和参数RDD求交集后返回一个新的RDD

distinct([numTasks]))

对源RDD进行去重后返回一个新的RDD

groupByKey([numTasks])

在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD

reduceByKey(func, [numTasks])

在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置

aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])

先按分区聚合 再总的聚合   每次要跟初始值交流 例如：aggregateByKey(0)(_+_,_+_) 对k/y的RDD进行操作

sortByKey([ascending], [numTasks])

在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD

sortBy(func,[ascending], [numTasks])

与sortByKey类似，但是更灵活 第一个参数是根据什么排序  第二个是怎么排序 false倒序   第三个排序后分区数  默认与原RDD一样

join(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD  相当于内连接(求交集)

cogroup(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable,Iterable))类型的RDD

cartesian(otherDataset)

两个RDD的笛卡尔积  的成很多个K/V

pipe(command, [envVars])

调用外部程序

coalesce(numPartitions)

重新分区 第一个参数是要分多少区，第二个参数是否shuffle 默认false  少分区变多分区 true   多分区变少分区 false

repartition(numPartitions)

重新分区 必须shuffle  参数是要分多少区  少变多

repartitionAndSortWithinPartitions(partitioner)

重新分区+排序  比先分区再排序效率高  对K/V的RDD进行操作

foldByKey(zeroValue)(seqOp)

该函数用于K/V做折叠，合并处理 ，与aggregate类似   第一个括号的参数应用于每个V值  第二括号函数是聚合例如：_+_

combineByKey

合并相同的key的值 rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)

partitionBy(partitioner)

对RDD进行分区  partitioner是分区器 例如new HashPartition(2

cache

RDD缓存，可以避免重复计算从而减少时间，区别：cache内部调用了persist算子，cache默认就一个缓存级别MEMORY-ONLY ，而persist则可以选择缓存级别

persist

Subtract(rdd)

返回前rdd元素不在后rdd的rdd

leftOuterJoin

leftOuterJoin类似于SQL中的左外关联left outer join，返回结果以前面的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可。

rightOuterJoin

rightOuterJoin类似于SQL中的有外关联right outer join，返回结果以参数中的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可

subtractByKey

substractByKey和基本转换操作中的subtract类似只不过这里是针对K的，返回在主RDD中出现，并且不在otherRDD中出现的元素

(二)Action：直接触发计算

触发代码的运行，我们一段spark代码里面至少需要有一个action操作。

动作

含义

reduce(func)

通过func函数聚集RDD中的所有元素，这个功能必须是课交换且可并联的

collect()

在驱动程序中，以数组的形式返回数据集的所有元素

count()

返回RDD的元素个数

first()

返回RDD的第一个元素(类似于take(1))

take(n)

返回一个由数据集的前n个元素组成的数组

takeSample(withReplacement,num, [seed])

返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子

takeOrdered(n, [ordering])

saveAsTextFile(path)

将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本

saveAsSequenceFile(path)

将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。

saveAsObjectFile(path)

countByKey()

针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。

foreach(func)

在数据集的每一个元素上，运行函数func进行更新。

aggregate

先对分区进行操作，在总体操作

reduceByKeyLocally

lookup

top

fold

foreachPartition

(三)RDD算子示例

1.创建一个Int类型的RDD---parallelize

val rdd1 = sc.parallelize(Array(5,1,75,32,647,23,5))

2.对每个元素乘以2(map)，并且排序(sortBy)---Transformation延迟计算

val rdd2 = rdd1.map(_*2).sortBy(x=>x,true)　　第一个参数是匿名函数，第二个true表示升序，false降序

注意：x=>x是匿名函数，传参是x,省略了类型，因为从数据中可以知道是Int　　---　　(x:Int)=>{x}　　---　　x=>x 　　返回值是x　　根据返回值的大小进行排序

3.以数组的形式返回数据集的所有元素(collect)---Action立即执行

rdd2.collect

4.过滤掉大于100的数(filter)---Transformation延迟计算

val rdd3 = rdd2.filter(_<100)

5.flatMap嵌套展开，并执行内部函数---Transformation延迟计算

val rdd5 = rdd4.flatMap(_.split(" "))

6. 集合运算---Transformation延迟计算

并集：

val rdd8 = rdd6.union(rdd7)

交集：

val rdd9 = rdd6.intersection(rdd7)

7.分组操作groupByKey--- Transformation延迟计算

四：RDD缓存机制---Transformation延迟计算

RDD通过persist方法或cache方法可以将前面的计算结果缓存，但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中，并供后面重用。

(一)案例实验

在Spark shell中读入文件，并进行文件行的计数，然后我们对数据进行缓存后再执行计算，注意，这里执行了count之后才会触发计算，将数据缓存下来，当再次执行count的时候，就会用到缓存的结果

第二次执行效率有所提高

五：RDD宽依赖和窄依赖

(一)宽依赖和窄依赖概念

每组依赖中左边为父RDD，右边为子RDD

窄依赖：是指每个父RDD的一个Partition最多被子RDD的一个Partition所使用，例如map、filter、union等操作都会产生窄依赖；(独生子女)

宽依赖：是指一个父RDD的Partition会被多个子RDD的Partition所使用，例如groupByKey、reduceByKey、sortByKey等操作都会产生宽依赖；(超生)

需要特别说明的是对join操作有两种情况：

(1)图中左半部分join：如果两个RDD在进行join操作时，一个RDD的partition仅仅和另一个RDD中已知个数的Partition进行join，那么这种类型的join操作就是窄依赖，例如图1中左半部分的join操作(join with inputs co-partitioned)；---多个子RDD，对于每个子RDD都满足窄依赖

(2)图中右半部分join：其它情况的join操作就是宽依赖,例如图1中右半部分的join操作(join with inputs not co-partitioned)，由于是需要父RDD的所有partition进行join的转换，这就涉及到了shuffle，因此这种类型的join操作也是宽依赖。

总之：没有依赖关系的stage是可以并行执行的，DAG根据宽依赖来划分stage，每个宽依赖的处理均会是一个stage的划分点。同一个stage中的多个操作会在一个task中完成。因为子RDD的分区仅依赖于父RDD的一个分区，因此这些步骤可以串行执行。

(二)依赖关系下的数据流视图

在spark中，会根据RDD之间的依赖关系将DAG图(有向无环图)划分为不同的阶段，

对于窄依赖，由于partition依赖关系的确定性，partition的转换处理就可以在同一个线程里完成，窄依赖就被spark划分到同一个stage中，

而对于宽依赖，只能等父RDD shuffle处理完成后，下一个stage才能开始接下来的计算。

因此spark划分stage的整体思路是：

从左往右推，遇到宽依赖就断开，划分为一个stage；

遇到窄依赖就将这个RDD加入该stage中。

因此在图2中RDD C,RDD D,RDD E,RDDF被构建在一个stage中,RDD A被构建在一个单独的Stage中,而RDD B和RDD G又被构建在同一个stage中。

在spark中，Task的类型分为2种：ShuffleMapTask和ResultTask；

简单来说，DAG的最后一个阶段会为每个结果的partition生成一个ResultTask，即每个Stage里面的Task的数量是由该Stage中最后一个RDD的Partition的数量所决定的！而其余所有阶段都会生成ShuffleMapTask；之所以称之为ShuffleMapTask是因为它需要将自己的计算结果通过shuffle到下一个stage中；也就是说上图中的stage1和stage2相当于mapreduce中的Mapper,而ResultTask所代表的stage3就相当于mapreduce中的reducer。

在之前动手操作了一个wordcount程序，因此可知，Hadoop中MapReduce操作中的Mapper和Reducer在spark中的基本等量算子是map和reduceByKey;不过区别在于：Hadoop中的MapReduce天生就是排序的；而reduceByKey只是根据Key进行reduce，但spark除了这两个算子还有其他的算子；因此从这个意义上来说，Spark比Hadoop的计算算子更为丰富。

六：RDD的Checkpoint(检查点)机制：容错机制

(一)检查点是辅助RDD的lineage(血统)进行容错的管理。

任务后面的步骤依赖于前面的步骤，如果一旦出现错误，则无法往后继续计算，就必须从头开始，这样必然会使得效率降低。整个计算的周期就叫lineage，lineage越长，出错的概率越大。所以我们可以设置一个检查点，如果后面的计算过程出错，则不需要从头重新进行计算，则可以从检查点的地方再次开始计算。

(二)RDD的检查点有两种类型

本地目录(即检查点的信息保存在本地的文件夹，一般用于开发和测试)，HDFS目录(保存在HDFS，多用于生产环境)

本地目录：需要把spark-shell运行在本地模式上

sc.setCheckpointDir("/root/temp/sparkcheckpoint")

HDFS目录：需要把spark-shell运行在集群模式上

sc.setCheckpointDir("hdfs://ns1/sparkcheckpoint")

七：Spark程序编写---WordCount

(一)配置eclipse环境

(二)使用scala语言编写---本地模式

package com.dt.spark

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

import org.apache.spark.rdd.RDDobjectWordCount {

def main(args:Array[String]):Unit={

val conf= newSparkConf()//设置运行模式为本地运行，不然默认是集群模式

conf.setMaster("local")//设置任务名

conf.setAppName("WordCount")//设置SparkContext,是SparkCore的程序入口

val sc = newSparkContext(conf)//读取文件，生成RDD

val file:RDD[String] = sc.textFile("E:\\storm-kafka\\a.txt")//将每一行数据进行嵌套展开、分割

val words:RDD[String] = file.flatMap(_.split(","))//进行单词映射

val wordOne:RDD[(String,Int)] = words.map((_,1))//进行单词计数

val wordCount:RDD[(String,Int)] = wordOne.reduceByKey(_+_)//按照单词出现次数，降序排序

val wordSort:RDD[(String,Int)] = wordCount.sortBy(x=>x._2, false)//将结果保存//wordSort.foreach(wordNumberPair => println(wordNumberPair._1 + " : " +wordNumberPair._2))//在命令行中打印该结果

wordSort.saveAsTextFile("E:\\Output")

sc.stop()

}

}

(二)使用scala语言编写---集群模式

package com.dt.spark

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

import org.apache.spark.rdd.RDDobjectWordCount {

def main(args:Array[String]):Unit={

val conf= newSparkConf()//设置运行模式为本地运行，不然默认是集群模式//conf.setMaster("local")//默认是集群模式　　master，是一个Spark、Mesos或者Yarn集群的URL，或者是local[*]；//设置任务名

conf.setAppName("WordCount")　　//appName，是用来在Spark UI上显示的应用名称；//设置SparkContext,是SparkCore的程序入口

val sc = newSparkContext(conf)//读取文件，生成RDD

val file:RDD[String] = sc.textFile("/spark/input/c.txt")　　//hdfs文件系统//将每一行数据进行嵌套展开、分割

val words:RDD[String] = file.flatMap(_.split(" "))//进行单词映射

val wordOne:RDD[(String,Int)] = words.map((_,1))//进行单词计数

val wordCount:RDD[(String,Int)] = wordOne.reduceByKey(_+_)//按照单词出现次数，降序排序

val wordSort:RDD[(String,Int)] = wordCount.sortBy(x=>x._2, false)//将结果显示

wordSort.collect().foreach(wordNumberPair => println(wordNumberPair._1 + ":" +wordNumberPair._2))//在命令行中打印该结果//wordSort.saveAsTextFile("E:\\Output")

sc.stop()

}

}

导出jar包，进行job提交。需要指定url

[hadoop@hadoopH1 ~]$ spark-submit --class com.dt.spark.WordCount --master spark://hadoopH2:7077 --executor-memory 500m --total-executor-cores 1 ./wc.jar

(三)使用Java7编写

package cn.spark.wc;

import java.util.Arrays;

import java.util.Iterator;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;public classWordCount {public static voidmain(String[] args) {//初始化sparkcontext

SparkConf conf = newSparkConf();

conf.setMaster("local").setAppName("WordCount");

JavaSparkContext sc= newJavaSparkContext(conf);//读取文件,生成RDD

JavaRDD fileRDD = sc.textFile("E:\\storm-kafka\\a.txt");//将每一行数据按照","分割

JavaRDD wordRDD = fileRDD.flatMap(new FlatMapFunction() {

@Overridepublic Iteratorcall(String line) throws Exception {//TODO Auto-generated method stub

return Arrays.asList(line.split(",")).iterator(); //返回列表迭代器

}

});//进行数据映射

JavaPairRDD wordOneRDD = wordRDD.mapToPair(new PairFunction() { //第一个是传入参数：单词，后面两个参数：是返回的元组参数

@Overridepublic Tuple2call(String word) throws Exception {//TODO Auto-generated method stub

return new Tuple2<>(word,1);

}

});//进行reduce操作

JavaPairRDD wordCountRDD = wordOneRDD.reduceByKey(new Function2() { //前两个是传入参数，后一个类型是返回参数

@OverridepublicInteger call(Integer i1, Integer i2) throws Exception {//TODO Auto-generated method stub

return i1 +i2;

}

});//下面进行排序操作sortByKey，由于是按照key进行排序，所以我们需要先将原来的元组，变为。---使用maptopair

JavaPairRDD countWordRDD = wordCountRDD.mapToPair(new PairFunction, Integer, String>() {

@Overridepublic Tuple2 call(Tuple2t) throws Exception {//TODO Auto-generated method stub

return new Tuple2<>(t._2,t._1);

}

});//开始使用sortByKey进行排序

JavaPairRDD sortCWRDD = countWordRDD.sortByKey(false); //默认升序，false降序//将我们处理的数据从转换为

JavaPairRDD resultRDD = sortCWRDD.mapToPair(new PairFunction, String, Integer>() {

@Overridepublic Tuple2 call(Tuple2t) throws Exception {//TODO Auto-generated method stub

return new Tuple2<>(t._2,t._1);

}

});//结果输出

resultRDD.saveAsTextFile("E:\\storm-kafka\\res");

}

}

补充：运行常见错误(一)无法运行，出现(launch error)

配置项目运行时参数：

-Dspark.master=local

补充：运行常见错误(二)无法运行，出现(unsupported major.minor version 52.0)运行环境错误

项目编译得到的class文件的版本高于运行环境中jre的版本号，高版本JDK编译的class不能在低版本的jvm虚拟机下运行，否则就会报这类错，因此无法运行！49，50，51，52是Java编译器内部的版本号，版本对应信息如下：

Unsupported major.minor version52.0 对应于 JDK1.8(JRE1.8)

Unsupported major.minor version51.0 对应于 JDK1.7(JRE1.7)

Unsupported major.minor version50.0 对应于 JDK1.6(JRE1.6)

Unsupported major.minor version49.0 对应于 JDK1.5(JRE1.5)

因此出现问题的原因就是：编译产生的class文件是jdk1.8版本的，而jvm环境低于1.8，因此报错：

修改Java运行环境：

运行结果查看：

(四)使用Java8编写---lambda表达式

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import scala.Tuple2;

import java.util.Arrays;public classSparkWordCountWithJava8 {public static voidmain(String[] args) {

SparkConf conf= newSparkConf();

conf.setAppName("WortCount");

conf.setMaster("local");

JavaSparkContext sc= newJavaSparkContext(conf);

JavaRDD fileRDD = sc.textFile("E:\\storm-kafka\\a.txt");

JavaRDD wordRdd = fileRDD.flatMap(line -> Arrays.asList(line.split(",")).iterator());

JavaPairRDD wordOneRDD = wordRdd.mapToPair(word -> new Tuple2<>(word, 1));

JavaPairRDD wordCountRDD = wordOneRDD.reduceByKey((x, y) -> x +y);

JavaPairRDD count2WordRDD = wordCountRDD.mapToPair(tuple -> new Tuple2<>(tuple._2, tuple._1));

JavaPairRDD sortRDD = count2WordRDD.sortByKey(false);

JavaPairRDD resultRDD = sortRDD.mapToPair(tuple -> new Tuple2<>(tuple._2, tuple._1));

resultRDD.saveAsTextFile("E:\\storm-kafka\\res");

}