spark-core入门:算子,Spark任务执行流程

作者: Java程序员YY | 来源:发表于2019-08-15 10:20 被阅读12次

    一、Spark Core概念讲解

    Spark Core:内核,也是Spark中最重要的部分。

    相当于Mapreduce

    SparkCore 和 Mapreduce都是进行离线数据分析

    SparkCore的核心:RDD(弹性分布式数据集),由分区组成

    二、RDD的五大特性

    1、RDD是由一系列的partition组成的。

    partition一般有三种方式产生

    (1)从Scala集合中创建,通过调用SparkContext#makeRDD或SparkContext#parallelize

    是可以指定partition 个数的,若指定了具体值,那么partition的个数就等于该值

    对于从 Scala集合中转换而来的RDD:默认的partition数为defaultParallelism,该值在不同的部署模式下不同:

    Local 模式:本机 cpu cores 的数量

    Mesos 模式:8

    Yarn:max(2, 所有 executors 的 cpu cores 个数总和)

    对于从外部数据加载而来的 RDD:默认的 partition 数为 min(defaultParallelism, 2)

    对于执行转换操作而得到的 RDD:视具体操作而定,如 map 得到的 RDD 的 partition 数与 父 RDD 相同;

    union 得到的 RDD的 partition 数为父 RDDs 的 partition 数之和...

    (2) 加载外部数据来创建 RDD,例如从 HDFS 文件、mysql 数据库读取数据等

    (3)由其它 RDD 执行 transform 操作转换而来

    2、函数是作用在每一个partition(split)上的。

    partition 是 RDD 的数据单位,代表了一个分区的数据。但这里千万不要搞错了,partition 是逻辑概念,

    是代表了一个分片的数据,而不是包含或持有一个分片的数据。

    3、RDD之间有一些列的依赖关系。

    如下图,可以看到RDD1产生RDD2;RDD2产生RDD3,即RDD分裂时返回的仍是RDD集,

    这样就形成了RDD间的层级依赖

    4、partitioner是作用在K,V格式的RDD上。

    partitioner即分区器,即决定RDD的每一条消息应该分到哪个分区。

    但只有 k, v 类型的RDD才能有partitioner(当然,非 key,value类型的RDD的partitioner为None)。

    5、RDD提供一系列最佳的计算位置

    spark提供了一种机制,在集群中运行任务时,会把task发送到集群中需要读取存储数据的节点来运行,不会将task随意发送到无关节点,这样减少数据节点间网络传输时间,做到计算移动、而数据不移动,实现了最佳计算位置。

    RDD特性理解图:

    三、 Spark任务执行流程

    1、基于standalone-client模式

    (1)spark集群启动,Worker节点向Master节点汇报各自资源,master掌握集群资源

    (2)在client提交spark应用程序,此时会再client启动一个Driver进程

    (3)Driver向Master请求资源,Master收到请求后,在集群中找到符合要求的节点启动Executor

    (4)Driver发送task到Worker,Worker上启动的Executor开始执行任务,完成后Driver负责并回收结果

    2、基于standalone-cluster模式

    standalone-cluster和standalone-client不同的是,client提交任务后,不会再启动Driver,而是Master在集群中找一台节点,即Worker上启动Driver,把Driver分散在集群节点中运行,这样避免了client上提交多个任务而引起的网络流量激增问题。client没有Driver 进程,也看不到task执行情况,以及接收不到输出结果。

    3、基于yarn-client模式

    yarn-cluste和yarn-client不同的是,client提交任务后,不会再启动Driver,而是RM在集群中找一台节点,即NM上启动Driver,把Driver分散在集群节点中运行,这样避免了client上提交多个任务而引起的网络流量激增问题。client上没有Driver 进程,也看不到task执行情况,以及接收不到输出结果。

    术语解释

    四、Spark代码流程

    (1)创建SparkConf对象

    可以设置Applicationname、可以设置运行模式及资源需求。

    (2)创建SparkContext对象

    (3) 基于Spark的上下文创建一个RDD,对RDD进行处理

    (4)应用程序中要有Action类算子来触发Transformation类算子执行。

    (5)关闭Spark上下文对象SparkContext。

    五、算子操作

    1、Transformations转换算子

    Transformations类算子是一类算子(函数)叫做转换算子,如map,flatMap,reduceByKey等。Transformations算子是延迟执行,也叫懒加载执行。

    2、Action行动算子

    Action类算子也是一类算子(函数)叫做行动算子,如foreach,collect,count等。Transformations类算子是延迟执行,Action类算子是触发执行。一个application应用程序中有几个Action类算子执行,就有几个job运行。

    3、持久化算子

    有时候会多次对同一个RDD进行计算,如果每次只是简单地调用行动算子,spark每次都会重新计算RDD及其所有依赖,这在迭代算法中消耗格外大。为了避免多次计算同一个RDD,spark引入了持久化 算子。

    (1)cache

    默认将RDD的数据持久化到内存中,cache是懒执行。看spark源码可以知道

    chche () = persist()=persist(StorageLevel.Memory_Only),相当于persist的StorageLevel.Memory_Only

    (2)persist

    可以指定持久化的级别,最常用的是MEMORY_ONLY和MEMORY_AND_DISK,persist是懒执行

    (3)checkpoint

    checkpoint将RDD持久化到磁盘,还可以切断RDD之间的依赖关系

    执行流程:当RDD的job执行完毕后,会从finalRDD从后往前回溯。当回溯到某一个RDD调用了checkpoint方法,会对当前的RDD做一个标记。Spark框架会自动启动一个新的job,重新计算这个RDD的数据,将数据持久化到指定的存储位置上

    优化:对RDD执行checkpoint之前,最好对这个RDD先执行cache,这样新启动的job只需要将内存中的数据拷贝到HDFS上就可以,省去了重新计算这一步

    最新免费java,架构,大数据AI编程资料获取添加

    薇信:18410263200

    通过验证填写“111”(备注必填)

    相关文章

      网友评论

        本文标题:spark-core入门:算子,Spark任务执行流程

        本文链接:https://www.haomeiwen.com/subject/grpajctx.html