spark中RDD的基本操作方式

spark中RDD的基本操作方式

读取本地文件

val rdd = sc.textFile("file:///home/tom/a.txt")
rdd.collect
val rdd1=rdd.map(s=>Integer.parseInt(s)+1)
rdd1.collect
rdd1.saveAsTextFile("file:///home/tom/result")

Spark是支持使用任何Hadoop支持的存储系统上的文件创建RDD的，比如说HDFS、Cassandra、HBase以及本地文件。通过调用SparkContext的textFile()方法，可以针对本地文件或HDFS文件创建RDD。

有几个事项是需要注意的：

1、如果是针对本地文件的话，如果是在windows上本地测试，windows上有一份文件即可；如果是在spark集群上针对linux本地文件，那么需要将文件拷贝到所有worker节点上。

2、Spark的textFile()方法支持针对目录、压缩文件以及通配符进行RDD创建。

3、Spark默认会为hdfs文件的每一个block创建一个partition，但是也可以通过textFile()的第二个参数手动设置分区数量，只能比block数量多，不能比block数量少。

读取HDFS文件

val rdd = sc.textFile("/input/a.txt")  //hdfs路径
rdd.collect

Spark的textFile()除了可以针对上述几种普通的文件创建RDD之外，还有一些特列的方法来创建RDD：

1、SparkContext.wholeTextFiles()方法，可以针对一个目录中的大量小文件，返回filename, fileContent组成的pair，作为一个PairRDD，而不是普通的RDD。普通的textFile()返回的RDD中，每个元素就是文件中的一行文本。

2、SparkContext.sequenceFileK, V方法，可以针对SequenceFile创建RDD，K和V泛型类型就是SequenceFile的key和value的类型。K和V要求必须是Hadoop的序列化类型，比如IntWritable、Text等。

3、SparkContext.hadoopRDD()方法，对于Hadoop的自定义输入类型，可以创建RDD。该方法接收JobConf、InputFormatClass、Key和Value的Class。

4、SparkContext.objectFile()方法，可以针对之前调用RDD.saveAsObjectFile()创建的对象序列化的文件，反序列化文件中的数据，并创建一个RDD。

基于集合创建rdd

val arr=Array(1,2,3,4,5,6)
val rdd=sc.parallelize(arr)
rdd.collect

val arr = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val rdd = sc.parallelize(arr)
val sum = rdd.reduce(_ + _)

调用parallelize()时，有一个重要的参数可以指定，就是要将集合切分成多少个partition。Spark会为每一个partition运行一个task来进行处理。Spark官方的建议是，为集群中的每个CPU创建2~4个partition。Spark默认会根据集群的情况来设置partition的数量。但是也可以在调用parallelize()方法时，传入第二个参数，来设置RDD的partition数量。比如parallelize(arr, 10)

rdd的常用函数

val rdd = sc.textFile("/input/a.txt")

rdd.collect  //返回所有数据
rdd.count
rdd.first    //注：没有last方法
rdd.take(3)  //取出rdd中的前三个元素
rdd.top(3)   //按照从大到小的顺序，取出最大的三个元素

foreach

rdd.foreach(s=>println(s))  //结果会夹杂在日志信息中
rdd.foreach(println)        //注意，或许会和上面的输出顺序不一致

map和flatMap区别

val rdd0=sc.parallelize(List(1,2,3,4))
rdd0.collect
rdd0.reduce((x,y)=>x+y)

val rdd1=rdd0.map(x=>x+1)
rdd1.collect
val rdd2=rdd0.map(x=>x.to(3))
rdd2.collect  //结果为4个集合
val rdd3=rdd0.flatMap(x=>x.to(3))
rdd3.collect  //结果合并为1个集合