zoukankan      html  css  js  c++  java
  • 大数据学习之Spark RDD算子45

    1:什么是RDD?

    RDD(Resilient Distributed DataSet)是分布式数据集。RDDSpark最基本的

    数据的抽象。

    scala中的集合。RDD相当于一个不可变、可分区、里面的元素可以并行计算的集合。

    RDD特点:具有数据流模型的特点

    自动容错

    位置感知调度

    可伸缩性

    RDD允许用户在执行多个查询时将工作集缓存在内存中,可以重用工作集,大大的提升了

    查询速度。

    2:RDD的属性(源码中的一段话)

    ² 一组分片(Partition,即数据集的基本组成单位。对于RDD来说,每个分片都会被一个计算任务处理,并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数,如果没有指定,那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。

    ² 一个计算每个分区的函数SparkRDD的计算是以分片为单位的,每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合,不需要保存每次计算的结果。

    ² RDD之间的依赖关系RDD的每次转换都会生成一个新的RDD,所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时,Spark可以通过这个依赖关系重新计算丢失的分区数据,而不是对RDD的所有分区进行重新计算。

    ² 一个Partitioner,即RDD的分片函数。当前Spark中实现了两种类型的分片函数,一个是基于哈希的HashPartitioner,另外一个是基于范围的RangePartitioner。只有对于于key-valueRDD,才会有Partitioner,非key-valueRDDParititioner的值是NonePartitioner函数不但决定了RDD本身的分片数量,也决定了parent RDD Shuffle输出时的分片数量。

    ² 一个列表,存储存取每个Partition的优先位置(preferred location)。对于一个HDFS文件来说,这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念,Spark在进行任务调度的时候,会尽可能地将计算任务分配到其所要处理数据块的存储位置。

    3:RDD类型分为:

    1Transformation

    转换

    2Action

    动作

    4:RDD的创建方式

    1通过外部的数据文件创建,如HDFS

    val rdd1 = sc.textFile(“hdfs://192.168.88.111:9000/data/data.txt”)

    2通过sc.parallelize进行创建

    val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))

    5:常用Transformation

    RDD中的所有转换都是延迟加载的,也就是说,它们并不会直接计算结果。相反的,它们只是记住这些应用到基础数据集(例如一个文件)上的转换动作。只有当发生一个要求返回结果给Driver的动作时,这些转换才会真正运行。这种设计让Spark更加有效率地运行。

    map(func)

    flatMap(func)

    sortby

    reduceByKey

    -filter

    过滤

     

    -union

    并集

     

    -intersection

    交集

    -groupByKey

    分组

    注意:这里结果顺序不一样。真好用到了并行处理。那个work先处理就先放回结果!

    -join

    关联

    -leftOuterJoin

    左连接

    保留左侧RDD,右侧如果join上保留没joinNone

    -rightOuterJoin

    右连接

    -cartesian

    笛卡尔积

    Transformation特点:

    1)生成新的RDD

    2lazy懒加载 等待处理

    3)并不会存储真正的数据,记录了转换关系

    6:常用的Action

    1)collect

    收集

    1)saveAsTextFile(path)

    真正处理任务的是Work,所以具体数据是存在bigdata113bigdata114机器上的

    存储文件

    三份数据:5B 5B 600B

    理想切分:5+5+600=610 610/3 = 203

    5B一片

    5B一片

    203一片

    203一片

    203一片

    1一片

    rdd1.partitions.length

    查看分区数

    3count

    计数

    4reduce

    聚合(这里我指定了分区数为2,也就是并行度)

    5countByKey()

    根据key计数

    6take(n)

    取出多少个元素

    7first

    返回RDD的第一个元素

    8takeOrdered(n)

    取出多少个元素 默认正序

    9top(n)

    倒序排序 取出元素

  • 相关阅读:
    在日志中记录Java异常信息的正确姿势
    基于Spring Boot架构的前后端完全分离项目API路径问题
    Spring生态简介
    WebSocket协议入门介绍
    Spring Boot程序正确停止的姿势
    python 中 __init__方法
    python中的if __name__ == 'main'
    python 类和实例
    内建模块 datetime使用
    内建模块collections的使用
  • 原文地址:https://www.cnblogs.com/hidamowang/p/11059916.html
Copyright © 2011-2022 走看看