spark

Posted 2020-08-29 伊米伊念

 

    /*
     * spark算子是RDD中定义的函数，可以对RDD中的数据进行转换和操作。
     * spark算子的作用：
     * 1.输入：在spark程序运行中，数据从外部数据空间（如分布式存储：textFile读取HDFS等，parallelize方法输入scala集合或数据）输入spark，
     *        数据进入spark运行时数据空间，转化为spark中的数据块，通过blockmanager进行管理
     * 2.运行：在spark数据输入形成RDD后便可以通过变换算子，如fliter等，对数据进行操作并将RDD转化为新的RDD，通过action算子，触发spark提交作业。
     *        如果数据需要复用，可以通过cache算子，将数据缓存到内存
     * 3.输出：程序运行结束数据会输出spark运行时控件，存储到分布式存储中（如savaAsTextFile输出到HDFS），
     *        或scala数据或集合中（collect输出到scala集合，count返回scala int类型）
     *        
     * spark算子大致上可以分三大类算子：
     * 1.value数据类型的transformation算子，这种变换不触发提交作业，针对处理的数据项是value型的数据
     *     map(func)  将原来RDD的每个数据项，使用map中用户自定义的函数func进行映射，转变为一个新的元素，并返回一个新的RDD。
     *     flatMap(func)  类似于map，但是输入数据项可以被映射到0个或多个输出数据集合中，所以函数func的返回值是一个数据项集合而不是一个单一的数据项。
     *     mapPartitions(func)  类似于map，但是该操作是在每个分区上分别执行，所以当操作一个类型为T的RDD时func的格式必须是Iterator<T> => Iterator<U>。即mapPartitions需要获取到每个分区的迭代器，在函数中通过这个分区的迭代器对整个分区的元素进行操作。
     *     union(otherDataset)  返回原数据集和参数指定的数据集合并后的数据集。使用union函数时需要保证两个RDD元素的数据类型相同，返回的RDD数据类型和被合并的RDD元素数据类型相同。该操作不进行去重操作，返回的结果会保存所有元素。如果想去重，可以使用distinct（）。
     *     cartesian(otherDataset)  对类型为T和U的两个数据集进行操作，返回包含两个数据集所有元素对的（T,U）格式的数据集。即对两个RDD内的所有元素进行笛卡尔积操作。
     *     groupBy : 根据自定义的东东进行分组。groupBy是基本RDD就有的操作。
     *     groupByKey([numTasks])   操作（K,V）格式的数据集，返回 (K, Iterable)格式的数据集。
                   注意，如果分组是为了按key进行聚合操作（例如，计算sum、average)，此时使用reduceByKey或aggregateByKey计算效率会更高。
                   注意，默认情况下，并行情况取决于父RDD的分区数，但可以通过参数numTasks来设置任务数。
     *     filter(func)  使用函数func对原RDD中数据项进行过滤，将符合func中条件的数据项组成新的RDD返回。
     *     distinct([numTasks]))  将RDD中的元素进行去重操作。
     *     subtract  返回一个由只存在于第一个RDD中而不存在于第二个RDD中的所有元素组成的RDD。它也需要shuffle
     *     sample(withReplacement, fraction, seed)   对数据采样。用户可以设定是否有放回（withReplacement）、采样的百分比（fraction）、随机种子（seed）。
     *     takeSample(withReplacement,num, [seed])  对一个数据集随机抽样，返回一个包含num个随机抽样元素的数组，参数withReplacement指定是否有放回抽样，参数seed指定生成随机数的种子。
     *     cache, persist  持久化
     * 2.key-value数据类型的transformation算子，这种变换不触发提交作业，针对处理的数据项是key-value型的数据
     *     mapValues  对各个键值对的值进行映射。该操作会保留RDD的分区信息。
     *     combineByKey 返回与输入数据的类型不同的返回值
     *     reduceByKey  与reduce相当类似，它们都接收一个函数，并使用该函数对值进行合并。不同的是，reduceByKey是transformation操作，reduceByKey只是对键相同的值进行规约，并最终形成RDD[(K, V)]，而不像reduce那样返回单独一个“值”。
     *     partitionBy 对RDD进行分区,可以减少大量的shuffle.
     *     cogroup  可以对多个RDD进行连接、分组、甚至求键的交集。其他的连接操作都是基于cogroup实现的。
     *     join   对数据进行内连接，也即当两个键值对RDD中都存在对应键时才输出。当一个输入对应的某个键有多个值时，生成的键值对RDD会包含来自两个输入RDD的每一组相对应的记录，也即笛卡尔积。
     *     leftOutJoin  即左外连接，源RDD的每一个键都有对应的记录，第二个RDD的值可能缺失，因此用Option表示。
     *     rightOutJoin  即右外连接，与左外连接相反。
     *     fullOuterJoin  即全外连接，它是是左右外连接的并集。
     * 3.action算子，这类算子会触发sparkcontext提交作业
     *     foreach  对每个元素进行操作，并不会返回结果。
     *     foreachPartition : 基于分区的foreach操作，操作分区元素的迭代器，并不会返回结果。
     *     saveAsTextFile 用于将RDD写入文本文件。spark会将传入该函数的路径参数作为目录对待，默认情况下会在对应目录输出多个文件，这取决于并行度。
     *     saveAsObjectFile saveAsObjectFile用于将RDD中的元素序列化成对象，存储到文件中。
     *     collect  收集RDD的元素到driver节点，如果数据有序，那么collect得到的数据也会是有序的。大数据量最好不要使用RDD的collect，因为它会在本机上生成一个新的Array，以存储来自各个节点的所有数据，此时更好的办法是将数据存储在HDFS等分布式持久化层上。
     *     collectAsMap   将结果以Map的形式返回，以便查询。
     *     reduceByKeyLocally 该函数将RDD[K,V]中每个K对应的V值根据映射函数来运算，运算结果映射到一个Map[K,V]中，而不是RDD[K,V]。
     *     lookup   返回给定键对应的所有值。
     *     count  返回RDD元素个数。
     *     top   如果为元素定义了顺序，就可以使用top返回前几个元素。
     *     reduce   对RDD中所有元素进行规约，最终得到一个规约结果。reduce接收的规约函数要求其返回值类型与RDD中元素类型相同。
     *     fold  与reduce类似，不同的是，它接受一个“初始值”来作为每个分区第一次调用时的结果。fold同样要求规约函数返回值类型与RDD元素类型相同。
     *     aggregate 与reduce和fold类似，但它把我们从返回值类型必须与所操作的RDD元素类型相同的限制中解放出来。
     *     
     */
    
    /*
     * spark运行架构
     * 术语：
     *    application：spark application的概念和hadoop mapreduce中的类似，指的是用户编写的spark应用程序，
     *                 包含了一个driver功能的代码和分布在集群中多个节点上运行的executor代码。
     *    driver：spark中的driver即运行上述application的main()函数并且创建sparkcontext，其中创建sparkcontext的目的是为了准备spark应用程序的运行环境。
     *            在spark中由sparkcontext负责和clustermanager通信，进行资源的申请，任务的分配和监控等；当executor部分运行完毕后，driver负责将sparkcontext关闭。
     *            通常用sparkcontext代表driver。
     *    executor：application运行在worker节点上的一个进程，该进程负责运行task，并且负责将数据存在内存或者磁盘上，每个application都有各自独立的一批executor。
     *             在spark on yarn模式下，其进程名称为CoarseGrainedExecutorBackend，类似于hadoop mapreduce中的YarnChild。
     *             一个CoarseGrainedExecutorBackend进程有且仅有一个executor对象，它负责将Task包装成taskRunner，并从线程池中抽取一个空闲的线程运行Task。
     *             每个CoarseGrainedExecutorBackend能并行运行Task的数量就取决于分配给它的CPU的个数了
     *    Cluster Manager:指的是在集群上获取资源的外部服务，目前有：
     *                    Standalone：spark原生的资源管理，由Master负责资源的分配
     *                    Hadoop Yarn：由Yarn中的ResourceManager负责资源分分配
     *    Worker：集群中任何可以运行application代码的节点，类似于yarn中NodeManager节点。
     *           在Standalone模式中指的就是通过Slave文件配置的Worker节点，在spark on yarn模式中指的就是NodeManager节点
     *   作业(Job)：包含多个Task组成的并行计算，往往由spark action催生，一个Job包含多个RDD及作用于相应RDD上的各种Operation。
     *   阶段(Stage):每个Job会被拆分很多组Task，每组任务被称为Stage，也可以成TaskSet，一个作业分多个阶段。
     *   任务(Task):被送到某个executor上的工作任务。
     */
    
    
    /*
     * spark运行基本流程
     * 1.构建Spark Application的运行环境（启动SparkContext），SparkContext向资源管理器（可以是Standalone， Mesos或YARN）注册并申请运行Executor资源。
     * 2.资源管理器分配Executor资源并启动StandaloneExecutorBackend，Executor运行情况将随着心跳发送到资源管理器上。
     * 3.SparkContext构建成DAG图，将DAG图分解成Stage，并把TaskSet发送给TaskScheduler。
     *   Executor向SparkContext申请Task，Task Schedule将Task发放到Executor运行同时SparkContext将应用程序代码发放给Executor。
     * 4.Task在Executor上运行，运行完毕释放所有资源。
     * 
     * spark注重建立良好的生态系统，它不仅支持多种外部文件存储系统，提供了多种多样的集群运行模式。
     * 部署在单台机器上时，既可以用本地（local）模式运行，也可以使用伪分布式模式来运行；
     * 当以分布式集群部署的时候，可以根据自己集群的实际情况选择Standalone模式（Spark自带的模式）， YARN-Client模式或者YARN-Cluster模式。
     * Spark的各种运行模式虽然在启动方式，运行位置，调度策略上各有不同，但它们的目的基本都是一致的，就是在合适的位置安全可靠的根据用户的配置和Job的需要运行和管理Task。
     * Standalone模式：是Spark实现的资源调度框架，其主要的节点有Client节点，Master节点和Worker节点。
     *               其中Driver既可以运行的Master节点上的，也可以运行在本地Client端。当用spark-shell交互式工具提交Spark的Job时，Driver在Master节点上运行；
     *               当使用spark—submit工具提交Job或者在Eclipse，IDEA等开发平台上使用"new SparkConf.setManager("spark://master:7077")"方式运行Spark任务时，Driver是运行在本地Client端上的。
     *               运行过程如下：
     *                  1.SparkContext连接Master，向Master注册并申请资源（CPU Core和Memory）；
     *                  2.Master根据SparkContext的资源申请要求和Worker心跳周期内报告的信息决定在哪个Worker上分配资源，然后在该Worker上提取资源，然后启动StandaloneExecutorBackend；
     *                  3.StandaloneExecutorBackend向SparkContext注册；
     *                  4.SparkContext将APPlication代码发送给StandaloneExecutorBackend；
     *                    并且SparkContext解析Application代码，构建DAG图，并提交给DAG Schedule分解成Stage（当碰到Action操作时，就会催生Job，每个Job中含有一个或多个Stage， Stage一般在获取外部数据和shuffle之前产生），
     *                    然后以Stage（或者称为TaskSet）提交给Task Scheduler，Task Schedule负责将Task分配到相应的Worker，最后提交给StandaloneExecutorBackend执行；
     *                  5.StandaloneExecutorBackend会建立Executor线程池，开始执行Task，并向SparkContext报告，知道Task完成；
     *                  6.所有Task完成后，SparkContext向Master注销，释放资源；
     * Spark on YARN运行过程：YARN是一种统一资源管理机制，在其上面可以运行多套计算框架。目前的大数据技术世界，大多数公司除了使用Spark来进行数据计算，由于历史原因火车单方面业务的性能考虑二使用着其他的计算框架，
     *                     比如MapReduce，Storm等计算框架。Spark基于此种情况开发了Spark on YARN的运行模式，由于借助了YARN良好的弹性资源管理机制，不仅部署Application更加方便，
     *                     而且用户在YARN集群中运行的服务和Application的资源可完全隔离，更具实践应用价值的是YARN可以通过队列的方式，管理同时运行在集群中的多个服务。
     *                     Spark on YARN模式根据Driver在集群中的位置分为两种模式：一种是YARN-Client模式，另一种是YARN-Cluster（或称为YARN-Standalone模式）。
     *                     YARN框架流程：任何框架与YARN的结合，都必须遵循YARN的开发模式。在分析Spark on YARN的实现细节之前，有必要先分析一下YARN框架的一些基本原理。
     *                     YARN框架的基本运行流程图为：ResourceManager负责将集群的资源分配给各个应用使用，而资源分配和调度的基本单位是Container，其中封装了机器资源，
     *                                            如内存、CPU、磁盘和网络等，每个任务会被分配一个Container，该任务只能在该Container中执行，并使用该Container封装的资源。
     *                                            NodeManager是一个个的计算节点，主要负责启动Application所需的Container，监控资源（内存、CPU、磁盘和网络等）的使用情况并将之汇报给ResourceManager。
     *                                            ResourceManager与NodeManager共同组成整个数据计算框架，ApplicationMaster与具体的Application相关，主要负责同ResourceManager协商以获取合适的Container，
     *                                            并跟踪这些Container的状态和监控其进度。
     *                     Yarn-Client：Yarn-Client模式中，Driver在客户端本地运行，这种模式可以使得Spark Application和客户端进行交互，因为Driver在客户端，所以可以通过webUI访问Driver的状态，       
     *                                  默认是http://hadoop1:4040访问，而YARN通过http://hadoop1:8088访问。
     *                                  Yarn-Client工作流：
     *                                       1.Spark Yarn Client向YARN的ResourceManager申请启动Application Master；
     *                                         同时在SparkContent初始化中将创建DAGScheduler和TASKScheduler等，由于我们选择的是Yarn-Client模式，程序会选择YarnClientClusterScheduler和YarnClientSchedulerBackend；
     *                                       2.ResourceManager收到请求后，在集群上选择一个NodeManager，为该应用程序分配第一个Container，要求她在这个Container中启动应用程序的ApplicationMaster，
     *                                         与YARN-Cluster区别的是在该ApplicationMaster不运行SparkContext，只与SparkContext进行联系进行资源的分配；
     *                                       3.Client中SparkContext初始化完毕后，与Application建立通讯，向ResourceManager注册，根据任务信息向ResourceManager申请资源（Container）；
     *                                       4.一旦ApplicationMaster申请到资源（也就是Container）后，便于对应的NodeManager通讯，要求它在获得的Container中启动CoarseGrainedExecutorBackend，
     *                                         CoarseGrainedExecutorBackend启动后会向Client中的SparkContext注册并申请Task；
     *                                       5.Client中的SparkContext分配Task给CoarseGrainedExecutorBackend执行，CoarseGrainedExecutorBackend运行Task并向Driver汇报运行的状态和进度，
     *                                         以让Client随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务；
     *                                       6.应用程序运行完成后，Client的SparkContext向ResourceManager申请注册并关闭自己；
     *                     Yarn-Cluster：在Yarn-Cluster模式中，当用户向Yarn中提交一个应用程序后，
     *                                   Yarn将分两个阶段运行该应用程序：
     *                                      1.把Spark的Driver作为一个ApplicationMaster在YARN集群中先启动
     *                                      2.由ApplicationMaster创建应用程序，然后为它向ResourceManager申请资源，并启动Executor来运行Task，同时监控它的整个运行过程，直到运行完成
     *                                   Yarn-Cluster工作流程：
     *                                      1.Spark Yarn Client向YARN中提交应用程序，包括ApplicationMaster程序，启动ApplicationMaster的命令，需要在Executor中运行的程序等；
     *                                      2.ResourceManager收到请求后，在集群中选择一个NodeManager，为该应用程序分配第一个Container，要求它在这个Container中启动ApplicationMaster，
     *                                        其中ApplicationMaster进行SparkContext等的初始化；
     *                                      3.ApplicationMaster向ResourceManager注册，这样用户可以直接通过ResourceManager查看应用程序的运行状态，
     *                                        然后它将采用轮询的方式通过RPC协议为各个任务申请资源，并监控它们的运行状态知道运行结束；
     *                                      4.一旦ApplicationMaster申请到资源（也就是Container）后，便于对应的NodeManager通信，要求它在获得的Container中启动CoarseGrainedExecutorBackend，
     *                                        CoarseGrainedExecutorBackend启动后会向ApplicationMaster中的SparkContext注册并申请Task。
     *                                        这一点和Standalone模式一样，只不过SparkContext在Spark Application中初始化时，使用CoarseGrainedExecutorBackend配合YarnClusterScheduler进行任务的调度，
     *                                        其中YarnClusterScheduler只是对TaskSchedulerImpl的一个简单包装，增加了对Executor的等待逻辑等；
     *                                      5.ApplicationMaster中对SparkContext分配Task给CoarseGrainedExecutorBackend执行，CoarseGrainedExecutorBackend运行Task并向ApplicationMaster汇报运行的状态和进度，
     *                                        以让ApplicationMaster随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务；
     *                                      6.应用程序运行完成后，ApplicationMaster向ResourceManager申请注册并关闭自己；
     *                     Yarn-Client和Yarn-Cluster的区别：理解Yarn-Client和Yarn-Cluster深层次的区别之前先清楚一个概念:ApplicationMaster。
     *                                                    在Yarn中，每个Application实例都有一个ApplicationMaster进程，它是Application启动的第一个容器。
     *                                                    它负责和ResourceManager打交道并请求资源，获取资源之后告诉NodeManager为其启动Container。
     *                                                    从深层次的含义讲Yarn-Cluster和Yarn-Client模式的区别其实就是ApplicationMaster进程的区别。
     *                                                    Yarn-Cluster模式下，Driver运行在ApplicationMaster中，负责向Yarn申请资源，并监督作业的运行状况。
     *                                                                      当用户提交作业之后，就可以关掉Client，作业会继续在Yarn上运行，因而Yarn-Cluster模式下不适合运行交互类型的作业。
     *                                                    Yarn-Client模式下，ApplicationMaster仅仅向Yarn请求Executor，Client会和请求的Container通信来调度他们的工作，也就是说Client不能离开。
     *                     
     */