1. Spark RDD 创建操作

1.1 数据集合

  parallelize 可以创建一个能够并行操作的RDD。其函数定义如下：

def parallelize[T: ClassTag](
    seq: Seq[T],
    numSlices: Int = defaultParallelism): RDD[T] = withScope {
  assertNotStopped()
  new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())
}

由定义可见有两个参数，第一个参数指定数据集合，第二个参数指定数据分区。

实例：由普通数组创建RDD 

1.2 外部数据源

textFiles 可以通过Hadoop支持的外部数据源(包括本地文件系统、HDFS、Cassandra、HBase等）建立RDD。其定义如下：

def textFile(
    path: String,
    minPartitions: Int = defaultMinPartitions): RDD[String]

第一个参数指定数据路径，第二个参数指定数据分区。

def defaultMinPartitions: Int = math.min(defaultParallelism, 2)

scala> sc.defaultParallelism

res0: Int = 2

由以上可知，如果第二个参数如果不设置默认为2，默认的并行度最大不超过2.

 实例1：读取本地文件创建RDD

 实例2：读取hdfs上的文件创建RDD 

2. Spark RDD 转换操作

  2.1 map(func)

  2.2 filter(func)

对RDD元素进行过滤，返回值为true的元素组成的一个新的数据集。

实例：返回数据集中的偶数

2.3 flatMap(func) 

2.4 mapPartitions(func)

def mapPartitions[U: ClassTag](
    f: Iterator[T] => Iterator[U],
 preservesPartitioning: Boolean = false): RDD[U]

private[spark] def mapPartitionsWithIndexInternal[U: ClassTag](
    f: (Int, Iterator[T]) => Iterator[U],
    preservesPartitioning: Boolean = false): RDD[U]

rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24 

 2.6 sample(withReplacement, fraction, seed)

def sample(
    withReplacement: Boolean,
    fraction: Double,
    seed: Long = Utils.random.nextLong): RDD[T]

withReplacement取回的数据是否放回抽样，fraction：比例，0.1表示10%,seed：随机种子，相同的seed得到的随机序列一样。

实例：

 2.7 union(otherDataset)

两个数据集合并，不去重，返回一个新的数据集，即所有旧的rdd的partition,直接移到新的rdd,新rdd 的Partition数量为旧rdd的partition数量的和。

实例：

查看新rdd的partition数量

scala> rdd3.partitions.length

res3: Int = 6

 2.8 intersection(otherDataset)

数据交集，相交的数据组成一个新的数据集返回。

实例：

2.9 distinct([numPartitions]))

去除两个数据集的重复数据，返回去重后的数据集。

实例：

2.10 groupByKey([numPartitions]))

分组操作，在一个由(K,V)对组成的数据集上调用，返回一个（K,Iterable[V])对的数据集。

def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])] 

实例：

 2.11 reduceByKey(func, [numPartitions])

分组聚合操作， 在一个(K,V)对的数据集上使用，返回一个（K,V)对的数据集。K相同的值，都被使用相同的reduce函数聚合在一起。

实例：相同的key的值加起来

2.12 aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions])

def aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner)(seqOp: (U, V) => U,combOp: (U, U) => U): RDD[(K, U)]  

分为两个分区，(1,3),(1，2）会落入0分区，(1,4),(2,3)落入1分区，每个分区分开计算。

实例2：

分为三个分区，(1,3)会落入0分区，(1,2)进入1分区，(1,4),(2,3)落入2分区，每个分区分开计算。

 2.13 combineByKey

对RDD中的数据集按照key进行聚合操作。聚合操作通过自定义函数提供。

def combineByKey[C](
    createCombiner: V => C,
    mergeValue: (C, V) => C,
    mergeCombiners: (C, C) => C): RDD[(K, C)] = self.withScope {
  combineByKeyWithClassTag(createCombiner, mergeValue, mergeCombiners)(null)
}

三个参数解释：

createCombiner：在遍历(k,v)时，如果combineByKey第一次遇到值为k的key（类型K),那么将这个(k,v)调用combineCombiner函数，将v转换为C.

mergeValue:在遍历(k,v)时，如果combineByKey不是第一次遇到值为k的key （类型K),那么将这个(k,v)调用mergeValue函数，它的作用是将v累加到聚合对象(类型为C）中。

mergeCombiners:combineByKey是在分布式环境下执行的，RDD的生个分区单独进行combineByKey操作，最后需要对各个分区的结果进行最后的聚合。

实例如下：

 2.14 sortByKey([ascending], [numPartitions])

按key对RDD进行排序，其定义如下：

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)
    : RDD[(K, V)] = self.withScope
{
  val part = new RangePartitioner(numPartitions, self, ascending)
  new ShuffledRDD[K, V, V](self, part)
    .setKeyOrdering(if (ascending) ordering else ordering.reverse)
}

实例：

2.15 join(otherDataset, [numPartitions])

连接操作，将输入数据集(K, V) 和另一数据集 (K, W)进行join, 返回两个集合匹配的(K, (V, W))集合对。即该操作过滤掉不匹配的key,然后返回相同K的V,W集合进行笛卡尔积操作。其定义如下：

def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))] = self.withScope {
  this.cogroup(other, partitioner).flatMapValues( pair =>
    for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, w)
  )
}

实例：

由此可见，不匹配的key(3,6)未显示。与sql类似，还有左连接、右连接及全连接操作函数：leftOuterJoin、rightOuterJoin及fullOuterJoin。

 2.16 cogroup(otherDataset, [numPartitions])

def cogroup[W](other: RDD[(K, W)], partitioner: Partitioner)
    : RDD[(K, (Iterable[V], Iterable[W]))]

实例：

def cartesian[U: ClassTag](other: RDD[U]): RDD[(T, U)] 

实例：

2.18 pipe(command, [envVars])

可以通过pipe使用shell命令来处理RDD

实例：

 2.19 coalesce(numPartitions)

coalesce(numPartitions: Int)将RDD进行重分区，默认只能减少分区，默认不进行shuffle，当开启shuffle时，可以扩大分区。其定义如下：

def coalesce(numPartitions: Int, shuffle: Boolean = false,
             partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
            (implicit ord: Ordering[T] = null)
    : RDD[T]

实例：减少分区

实例：扩大分区

实例：开启 shuffle后，可以扩大分区数

 2.20 repartition(numPartitions)

对RDD进行重分区，可以增加分区，也可减少分区，它创建新的分区，会进行shuffer操作

实例：

2.21 repartitionAndSortWithinPartitions(partitioner)

根据给定的分区程序对RDD进行重新分区，并在每个生成的分区内按键对记录进行排序。 这比调用重新分区，它要比使用repartition And sortByKey 效率高，这是由于它的排序是在shuffle过程中进行，一边shuffle，一边排序。定义如下：

def repartitionAndSortWithinPartitions(partitioner: Partitioner): RDD[(K, V)] = self.withScope {
  new ShuffledRDD[K, V, V](self, partitioner).setKeyOrdering(ordering)
}

使用repartitionAndSortWithinPartitions时，需要自己传入一个分区器，这个分区器可以使用系统提供的，也可以是自定义的，以下实例我们使用系统的HashPartitioner

实例：

3. 转换操作

3.1 reduce(func)

对数据集中每个元素执行指定的聚集函数（有两个输入参数，一个返回值） ，这个函数必须是可交换的和组合的。

实例：

3.4 first()

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

实例：

3.5 take(n)

返回集合中前n个元素的数组。

3.6 takeSample(withReplacement, num, [seed])

返回包含随机的num个元素的数组。第一个参数withReplacement是抽样时是否放回，第二个参数num会精确指定抽样数，而不是比例.第三个参数seed是随机种子。

实例：

res3: Array[Int] = Array(1, 2, 3) 

3.7  takeOrdered(n, [ordering])

按自然顺序或者自定义比较器返回第1到n元素的数组

实例：

res6: Array[Int] = Array(1, 2, 3, 4, 5)

 3.8 saveAsTextFile(path)

把数据集中的元素转换为文本文件写到指定的目录（本地系统、HDFS或者其它hadoop支持的文件系统).Spark将每个元素调用toString方法转换为文本文件中的一行。

实例：将数据转为文本存储到hdfs中。

 3.11 countByKey()

scala> val rdd=sc.parallelize(Array((1,2),(1,3),(1,5),(2,4),(2,6),(3,8)),3)

3.12 foreach(func)

对数据集中每个元素执行func函数。

参考文献：

http://spark.apache.org/docs/latest/rdd-programming-guide.html

Spark MLlib机器学习 -黄美灵

版权声明：本文为博主原创文章，转载请注明出处：https://www.cnblogs.com/abcdwxc/p/9867475.html