spark几个重要的概念区分

时间:2018-04-03作者:klpeng分类:IT综合浏览:13629评论:0

1、RDD和DStream的区别

RDD:

弹性数据集,其中包含了多个partition,每个子集partition可以分布在不同节点上,在进行处理时分别在不同机器上进行处理;

DStream:

对数据流按时间切分出来的一小批次,每个DStream对应多个RDD,这些RDD是按照时间维度进行划分的,
关系:相当于一整条数据流DStream被切分成了多个RDD,每个DStream对应多个RDD;一个RDD对应多个partition

2、foreach与foreachPartition的区别

foreach:源码:
def foreach(f: T => Unit): Unit = withScope {
val cleanF = sc.clean(f)
sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))}

获得每个partition中数据iterator 然后在遍历这个迭代器时对其中的每条数据进行入参function操作

foreachPartition:源码:
def foreachPartition(f: Iterator[T] => Unit): Unit = withScope {
val cleanF = sc.clean(f)
sc.runJob(this, (iter: Iterator[T]) => cleanF(iter))}

和foreach一样都是先获取partition中的数据iterator,不同的是在此方法中没有对iterator进行遍历,而是把这个iterator交给了入参function处理
总结:都是对partition中的数据进行操作,foreach是对每条信息,foreachpartition操作的是整个partition的iterator

3、map、mapPartitions和mapPartitionsWithIndex区别

map源码:
def map[U: ClassTag](f: T => U): RDD[U] = withScope {
val cleanF = sc.clean(f)
new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))}

这个方法的的入参函数的参数是rdd中的具体一条信息,这样操作会不停的创建对象,例如:
如果需要把结果写到Mysql中,哪来一条记录就得生成一个连接

mapPartitions源码:
def mapPartitions[U: ClassTag]( 
f: Iterator[T] => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U] = withScope {
val cleanedF = sc.clean(f)
new MapPartitionsRDD(
this,
(context: TaskContext, index: Int, iter: Iterator[T]) => cleanedF(iter),
preservesPartitioning)}

这个方法的的入参函数的参数是一个迭代器,对于一个RDD的每个分区进行操作,这样就可以对一个分区只建立一个连接

mapPartitionsWithIndex源码:
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U] = withScope {
val cleanedF = sc.clean(f)
new MapPartitionsRDD(
this,
(context: TaskContext, index: Int, iter: Iterator[T]) => cleanedF(index, iter),
preservesPartitioning)}

这个方法与mapPartition类似;只不过入参的函数的参数有两个:一个是分区号,一个是迭代器

4、Spark的高级排序和Spark的高级分区

排序:

spark中的排序是通过隐式转换实现的,源码:

implicit def rddToOrderedRDDFunctions[K : Ordering : ClassTag, 
                                V: ClassTag](rdd: RDD[(K, V)])
: OrderedRDDFunctions[K, V, (K, V)] = {
new OrderedRDDFunctions[K, V, (K, V)](rdd)}

其中OrderedRDDFunctions是SortByKey所在的类
再看sortBykey源码:

private val ordering = implicitly[Ordering[K]]
/**
    * Sort the RDD by key, so that each partition contains a sorted range of the elements. Calling
    * `collect` or `save` on the resulting RDD will return or output an ordered list of records
    * (in the `save` case, they will be written to multiple `part-X` files in the filesystem, in
    * order of the keys).
**/
    // TODO: this currently doesn't work on P other than Tuple2!
    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)}

在转换调用sortByKey方法时,会从上下文中提取Ordering[K]

private val ordering = implicitly[Ordering[K]]

其中

implicit def rddToOrderedRDDFunctions[K : Ordering : ClassTag, V: ClassTag](rdd: RDD[(K, V)])

这里的k:Ordering意思是指k必须是可以转换成ordering的子类
所以我们有两种方式

1、在定义K的排序时可以使用隐式值
implicit val po: Ordering[Monkey] = new Ordering[Monkey] {
override def compare(x: Monkey, y: Monkey): Int = { 
x.name.compareTo(y.name)
}}

ssc.textFileStream("").foreachRDD(rdd => rdd.map(msg => (Monkey(msg), msg)).sortByKey())
2、key类实现Ordered[k]接口中的compare方法
case class Monkey(name:String) extends Ordered[Monkey]{
    override def compare(that: Monkey): Int = {
    this.name.compare(that.name)
    } 
    def run(): Unit ={
    println("monkey run fast")
    }
}
private def test() { 
//implicit val po: Ordering[Monkey] = new Ordering[Monkey] {
ssc.textFileStream("").foreachRDD(rdd => rdd.map(msg => (Monkey(msg), msg)).sortByKey())
}
打赏
文章版权声明:除非注明,否则均为彭超的博客原创文章,转载或复制请以超链接形式并注明出处。
相关推荐

发表评论:

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。

猜你喜欢