def glom():RDD[Array[T]]
image.png

glom算子效果和flatMap(func)效果是相反的.

作用: 将RDD中每一个分区的元素合并成一个数组，形成新的 RDD 类型是RDD[Array[T]]
假如说我有个RDD1,RDD1里面有两个分区,两个分区分别放1 2 和 3 4. 经过glom算子之后就得到一个新的RDD2, RDD2还是两个分区,分区号没变,还是原来的分区号,此时每个分区的数据都合并成一个数组了, 数组里面是原来分区的值,.

比如说之前分区0里面的数据是 1 2 , 调用glom算子之后,分区0里面的数据是 array(1,2)了.

案例

  2. /**
  3.  * -将RDD一个分区中的元素，组合成一个新的数组
  4.  */
  5. object Spark04_Transformation_glom222 {
  6.   def main(args: Array[String]): Unit = {
  7.     //创建SparkConf并设置App名称
  8.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  10.     //创建SparkContext，该对象是提交Spark App的入口
  11.     val sc: SparkContext = new SparkContext(conf)
  12.     //2个分区
  13.     val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6), 2)
  14.     //给每个分区的元素的值变成数组
  15.     var rddglom = rdd.glom()
  16.     //
  17.     var mappartitionsRDD = rddglom.mapPartitionsWithIndex {
  18.       (index, datas) => {
  19.         datas
  20.       }
  21.     }
  23.     var array: Array[Array[Int]] = mappartitionsRDD.collect()
  24.     // 获取第一个分区的元素,是个数组,按逗号变成字符串
  25.     println(array(0).mkString(",")) //输出: 1,2,3
  26.     //获取第二个分区的元素,是个数组,按逗号变成字符串
  27.     println(array(1).mkString(",")) //输出: 4,5,6
  30.     // 关闭连接
  31.     sc.stop()
  32.   }
  33. }
  1.  import org.apache.spark.rdd.RDD
  2. import org.apache.spark.{SparkConf, SparkContext}
  4. /**
  5.  * -将RDD一个分区中的元素，组合成一个新的数组
  6.  */
  7. object Spark04_Transformation_glom {
  8.   def main(args: Array[String]): Unit = {
  9.     //创建SparkConf并设置App名称
  10.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  12.     //创建SparkContext，该对象是提交Spark App的入口
  13.     val sc: SparkContext = new SparkContext(conf)
  15.     val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6), 2)
  17.     //    以分区为单位输出每个分区的元素.
  18.     println("------------没有glom之前------------")
  19.     rdd.mapPartitionsWithIndex {
  20.       (index, datas) => {
  21.         println(index + "--->" + datas.mkString(","))//给数组展示出来.
  22.         datas
  23.       }
  24.     }.collect()
  26.     /*
  27.     0--->1,2,3
  28.     1--->4,5,6
  29.      */
  31.     println("------------调用glom之后------------")
  32.     val newRDD: RDD[Array[Int]] = rdd.glom()
  33.     newRDD.mapPartitionsWithIndex {
  34.       (index, datas) => {
  35.         println(index + "--->" + datas.next().mkString(","))
  36.         datas
  37.       }
  38.     }.collect()
  39.     /*
  40.     0--->1,2,3
  41.     1--->4,5,6
  42.     */
  44.     // 关闭连接
  45.     sc.stop()
  46.   }
  47. }
  4. import org.apache.spark.rdd.RDD
  5. import org.apache.spark.{SparkConf, SparkContext}
  7. object Spark04_Transformation_glom222 {
  8.   def main(args: Array[String]): Unit = {
  9.     val conf: SparkConf = new SparkConf().setAppName("Glom").setMaster("local[2]")
  10.     val sc: SparkContext = new SparkContext(conf)
  11.     val list1 = List(30, 50, 70, 60, 10, 20)
  12.     val rdd1: RDD[Int] = sc.parallelize(list1, 2)
  13.     //        val rdd2 = rdd1.glom().map(x => x.toList)
  14.     val rdd2 = rdd1.glom().map(_.toList)
  15.     // 一个分区组成一个数组
  16.     rdd2.collect.foreach(println)
  17.     //输出: List(30, 50, 70)
  18.     //List(60, 10, 20)
  19.     sc.stop()
  21.   }
  22. }