Spark【RDD编程（二）RDD编程基础】-阿里云开发者社区

前言

接上午的那一篇，下午我们学习剩下的RDD编程，RDD操作中的剩下的转换操作和行动操作，最好把剩下的RDD编程都学完。

RDD 转换操作

6、distinct

对 RDD 集合内部的元素进行去重，然后把去重后的其他元素放到一个新的 RDD 集合内。

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object RDDTransForm {
  def main(args: Array[String]): Unit = {
    // 创建SparkContext对象
    val conf = new SparkConf()
    conf.setAppName("spark core rdd transform").setMaster("local")
    val sc = new SparkContext(conf)
    // 通过并行集合创建RDD对象
    val arr = Array("Spark","Flink","Spark","Storm")
    val rdd1: RDD[String] = sc.parallelize(arr)
    val rdd2: RDD[String] = rdd1.distinct()
    rdd2.foreach(println)
    //关闭SparkContext
    sc.stop()
  }
}

运行输出：

Flink
Spark
Storm

可以看到，重复的元素"Spark"被去除掉。

7、union

对两个 RDD 集合进行并集运算，并返回新的 RDD集合，虽然是并集运算，但整个过程不会把重复的元素去除掉。

// 通过并行集合创建RDD对象
    val arr1 = Array("Spark","Flink","Storm")
    val arr2 = Array("Spark","Flink","Hadoop")
    val rdd1: RDD[String] = sc.parallelize(arr1)
    val rdd2: RDD[String] = sc.parallelize(arr2)
    val rdd3: RDD[String] = rdd1.union(rdd2)
    rdd3.foreach(println)

运行结果：

Spark
Flink
Storm
Spark
Flink
Hadoop

可以看到，重复的元素"Spark"和"Flink"没有被去除。

8、intersection

对两个RDD 集合进行交集运算。

// 通过并行集合创建RDD对象
    val arr1 = Array("Spark","Flink","Storm")
    val arr2 = Array("Spark","Flink","Hadoop")
    val rdd1: RDD[String] = sc.parallelize(arr1)
    val rdd2: RDD[String] = sc.parallelize(arr2)
    val rdd3: RDD[String] = rdd1.intersection(rdd2)
    rdd3.foreach(println)

运行结果：

Spark
Flink

"Spark"和"Flink"是两个RDD集合都有的。

9、subtract

对两个RDD 集合进行差集运算，并返回新的RDD 集合。

rdd1.substract(rdd2) 返回的是 rdd1有而rdd2中没有的元素，并不会把rdd2中有rdd1中没有的元素也包进来。

// 通过并行集合创建RDD对象
    val arr1 = Array("Spark","Flink","Storm")
    val arr2 = Array("Spark","Flink","Hadoop")
    val rdd1: RDD[String] = sc.parallelize(arr1)
    val rdd2: RDD[String] = sc.parallelize(arr2)
    val rdd3: RDD[String] = rdd1.subtract(rdd2)
    rdd3.foreach(println)

运算结果：

Storm

"Storm"是rdd1中有的二rdd2中没有的，并不会返回"Hadoop"。

10、zip

把两个 RDD 集合中的元素以键值对的形式进行合并，所以需要确保两个RDD 集合的元素个数必须是相同的。

// 通过并行集合创建RDD对象
    val arr1 = Array("Spark","Flink","Storm")
    val arr2 = Array(1,3,5)
    val rdd1: RDD[String] = sc.parallelize(arr1)
    val rdd2: RDD[Int] = sc.parallelize(arr2)
    val rdd3: RDD[(String,Int)] = rdd1.zip(rdd2)
    rdd3.foreach(println)

运行结果：

(Spark,1)
(Flink,3)
(Storm,5)

RDD 行动操作

RDD 的行动操作是真正触发计算的操作，计算过程十分简单。

1、count

返回 RDD 集合中的元素数量。

2、collect

以数组的形式返回 RDD 集合中所有元素。

3、first

返回 RDD 集合中的第一个元素。

4、take(n)

返回 RDD 集合中前n个元素。

5、reduce(func)

以规则函数func对RDD集合中的元素进行循环处理，比如将所有元素加到一起或乘起来。

6、foreach

对RDD 集合进行遍历，输出RDD集合中所有元素。

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object RDDAction {
  def main(args: Array[String]): Unit = {
    // 创建SparkContext对象
    val conf = new SparkConf()
    conf.setAppName("spark core rdd transform").setMaster("local")
    val sc = new SparkContext(conf)
    //通过并行集合创建 RDD 对象
    val arr: Array[Int] = Array(1,2,3,4,5)
    val rdd: RDD[Int] = sc.parallelize(arr)
    val size: Long = rdd.count()
    val nums: Array[Int] = rdd.collect()
    val value: Int = rdd.first()
    val res: Array[Int] = rdd.take(3)
    val sum: Int = rdd.reduce((v1, v2) => v1 + v2)
    println("size = " + size)
    println("The all elements are ")
    nums.foreach(println)
    println("The first element in rdd is " + value)
    println("The first three elements are ")
    res.foreach(println)
    println("sum is " + sum)
    rdd.foreach(print)
    //关闭SparkContext
    sc.stop()
  }
}

运行结果：

size = 5
The all elements are 
1
2
3
4
5
The first element in rdd is 1
The first three elements are 
1
2
3
sum is 15
12345
Process finished with exit code 0

文本长度计算案例

计算 data 目录下的文件字节数（文本总长度）。

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object FileLength {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("spark core rdd transform").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd1: RDD[String] = sc.textFile("data")
    val rdd2: RDD[Int] = rdd1.map(line => line.length)
    val fileLength: Int = rdd2.reduce((len1, len2) => len1 + len2)
    println("File length is " + fileLength)
    sc.stop()
  }
}

持久化

在Spark 中，RDD采用惰性机制，每次遇到行动操作，就会从头到尾开始执行计算，这对于迭代计算代价是很大的，因为迭代计算经常需要多次重复使用相同的一组数据。

使用cache() 方法将需要持久化的RDD对象持久化进缓存中
使用unpersist() 方法将持久化rdd从缓存中释放出来

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object RDDCache {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("spark core rdd transform").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List("Hadoop","Spark","Hive","Flink")
    val rdd: RDD[String] = sc.parallelize(list)
    rdd.cache()
    println(rdd.count())  //第一次行动操作
    println(rdd.collect.mkString(",")) //第二次行动操作
    rdd.unpersist() //把这个持久化的rdd从缓存中移除，释放内存空间
    sc.stop()
  }
}

分区

分区的作用

RDD 是弹性分布式数据集，通过 RDD 都很大，会被分成多个分区，分别保存在不同的节点上。进行分区的好处：

增加并行度。一个RDD不分区直接进行计算的话，不能充分利用分布式集群的计算优势；如果对RDD集合进行分区，由于一个文件保存在分布式系统中不同的机器节点上，可以就近利用本分区的机器进行计算，从而实现多个分区多节点同时计算，并行度更高。
减少通信开销。通过数据分区，对于一些特定的操作（如join、reduceByKey、groupByKey、leftOuterJoin等），可以大幅度降低网络传输。

分区的原则

使分区数量尽量等于集群中CPU核心数目。可以通过设置配置文件中的 spark.default.parallelism 这个参数的值，来配置默认的分区数目。

设置分区的个数

1、创建 RDD对象时指定分区的数量

1.1、通过本地文件系统或HDFS加载

sc.textFile(path,partitionNum)

1.2、通过并行集合加载

对于通过并行集合来创建的RDD 对象，如果没有在参数中指定分区数量，默认分区数目为 min(defaultParallelism,2) ，其中defaultParallelism就是配置文件中的spark.default.parallelism。如果是从HDFS中读取文件，则分区数目为文件分片的数目。

2、使用repartition()方法重新设置分区个数

val rdd2 = rdd1.repartition(1)    //重新设置分区为1

自定义分区函数

继承 org.apache.spark.Partitioner 这个类，并实现下面3个方法：

numPartitions: Int ，用于返回创建出来的分区数。
getPartition(key: Any)，用于返回给定键的分区编号(0~paratitionNum-1)。
equals()，Java中判断相等想的标准方法。

注意：Spark 的分区函数针对的是（key，value）类型的RDD，也就是说，RDD中的每个元素都是（key,value）类型的，然后函数根据 key 对RDD 元素进行分区。所以，当要对一些非（key，value）类型的 RDD 进行自定义分区时，需要首先把 RDD 元素转换为（key，value）类型，然后再使用分区函数。

案例

将奇数和偶数分开写到不同的文件中去。

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}
class MyPartitioner(numParts: Int = 2) extends Partitioner{
  //覆盖默认的分区数目
  override def numPartitions: Int = numParts
  //覆盖默认的分区规则
   override def getPartition(key: Any): Int = {
    if (key.toString.toInt%2==0) 1 else 0
  }
}
object MyPartitioner{
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("partitioner").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)
    val data: Array[Int] = (1 to 100).toArray
    val rdd: RDD[Int] = sc.parallelize(data,5)
    val savePath:String = System.getProperty("user.dir")+"/data/rdd/out"
    rdd.map((_,1)).partitionBy(new MyPartitioner()).map(_._1).saveAsTextFile(savePath)
    sc.stop()
  }
}

我们在代码中创建RDD 对象的时候，我们指定了分区默认的数量为 5，然后我们使用我们自定义的分区，观察会不会覆盖掉默认的分区数量：

运行结果：

我们可以看到，除了校验文件，一共生成了两个文件，其中一个保存了1~100的所有奇数，一个保存了1~100的所有偶数；

综合案例

在上一篇博客中，我们已经做过WordCount了，但是明显篇幅比较长，这里我们简化后只需要两行代码：

    //使用本地文件作为数据加载创建RDD 对象
    val rdd: RDD[String] = sc.textFile("data/word.txt")
    //RDD("Hadoop is good","Spark is better","Spark is fast")
    val res_rdd: RDD[(String,Int)] = rdd.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
    //flatMap:
    //RDD(Array("Hadoop is good"),Array("Spark is better"),Array("Spark is fast"))
    //RDD("Hadoop","is",good","Spark","is","better","Spark","is","fast"))

运行结果：

(Spark,2)
(is,3)
(fast,1)
(good,1)
(better,1)
(Hadoop,1)

总结

至此，我们RDD基础编程部分就结束了，但是RDD编程还没有结束，接下来我会继续学习键值对RDD、数据读写，最后总结性低做一个大的综合案例。

Spark【RDD编程（二）RDD编程基础】

前言

RDD 转换操作

6、distinct

7、union

8、intersection

9、subtract

10、zip

RDD 行动操作

1、count

2、collect

3、first

4、take(n)

5、reduce(func)

6、foreach

文本长度计算案例

持久化

分区

分区的作用

分区的原则

设置分区的个数

1、创建 RDD对象时指定分区的数量

1.1、通过本地文件系统或HDFS加载

1.2、通过并行集合加载

2、使用repartition()方法重新设置分区个数

自定义分区函数

案例

综合案例

总结

热门文章

最新文章

相关课程

相关电子书