1.Spark体系结构的核核心主要功能和组件
2.图计算用spark+scala+graphx进行图计算?
3.Spark ML系列RandomForestClassifier RandomForestClassificationModel随机森林原理示例源码分析
4.Spark Core读取ES的分区问题分析
5.Spark源码解析2-YarnCluster模式启动

Spark体系结构的主要功能和组件

       Spark已经成为全球主要行业中功能强大且需求量最大的大数据框架，其可访问性和强大功能使其能够处理大数据挑战。心源拥有超过,码深名成员的良好用户基础，以及多人对代码所做的入理贡献，它已成为阿里巴巴、思想亚马逊、源码图解kafka源码eBay、分析雅虎、核核心腾讯、心源百度等主流企业的码深首选框架。InMobi数据科学与市场高级副总裁Rajiv Bhat表示：“Spark使开发机器学习模型的入理时间从六到七个月缩短至每天约四个模型。”Spark作为开源框架，思想已成为Apache Software Foundation运行最频繁的源码项目，目前在大数据处理领域是分析市场领导者。

       Spark与Hadoop之间的核核心主要区别在于，Hadoop基于处理已存储一段时间的数据块的概念，而Spark则用于实时处理。Hadoop在年是大数据领域的突破性技术，但直到年Spark引入时才如此。Spark的主要销售主张是实时速度，因为它比Hadoop的MapReduce框架快倍。Spark功能包括实时数据处理的开放源代码计算集群框架，提供对具有内置并行性和容错性的整个集群进行编程的接口，其核心建立在Hadoop的MapReduce框架上并扩展到更多计算类型。

       Spark体系结构基于两个主要的抽象，包括主/从体系结构，具有一个Master和多个Slave/Worker。单个Java进程由驱动程序和执行程序运行，换钱源码系统用户可以在不同机器上运行它们，以适应垂直集群、混合机器配置或在同一水平的Spark集群。Spark体系结构包括驱动程序、执行程序和集群管理器的角色。驱动程序是Spark应用程序的中心点，负责将用户代码转换为实际的Spark作业。执行者主要负责执行任务，而集群管理器提供了不同的调度功能集，以分配和取消分配各种物理资源，如客户端Spark作业、CPU内存等。

       Spark应用程序的运行时架构涉及客户端提交的Spark应用程序代码被转换为逻辑DAG（有向无环图），进行各种优化，如对转换进行流水线处理，然后转换为具有一组阶段的物理执行计划。物理执行计划由各种小型物理执行单元组成，这些任务组合在一起并发送到Spark集群。驱动程序与集群管理器进行交互，进行资源协商，集群管理器在从属节点上启动执行程序。驱动程序根据数据放置将任务发送给集群管理器，执行程序在执行前向驱动程序注册，驱动程序在应用程序运行时进行监视。当驱动程序main（）方法退出或stop（）方法退出时，它将终止所有执行程序并将其从集群管理器中释放。

       因此，可用社区源码Spark架构因其易用性、可访问性以及处理大数据任务的能力，最终在Hadoop中占主导地位。它在许多行业中广泛应用，将Hadoop MapReduce提升至全新水平，在数据处理方面几乎没有任何改组。内存中的数据存储和实时数据处理提高了系统的效率倍，惰性评估有助于速度。关注微信公众号“海牛大数据”（ID：hainiudashuju），加入实战技术论坛，参与大数据技术交流社区。

图计算用spark+scala+graphx进行图计算?

       EdgeRDD在Spark GraphX中作为核心类之一，用于存储和处理图的边数据，它扩展了RDD[Edge[ED]]，通过列式格式在每个分区上存储边，以提高性能，并且可以额外存储与每条边关联的顶点属性，以提供三元组视图。

       EdgeRDD提供了一系列方法来操作和转换图的边数据，包括但不限于：

       mapValues：对边进行映射操作，将边属性转换为特定形式。

       reverse：反转所有边的方向。

       innerJoin：与另一个EdgeRDD进行内连接操作。

       这些方法允许进行映射、反转和连接等操作，以满足不同的图计算需求。

       示例中，静思文学源码我们首先创建了一个包含边的RDD，然后通过EdgeRDD创建了对象。接着，使用mapValues方法将边属性转换为大写形式。随后，reverse方法用于反转所有边的方向。最后，innerJoin方法将原始EdgeRDD与反转后的EdgeRDD进行内连接操作，将两个边的属性拼接为新的字符串，得到了最终的结果EdgeRDD对象。

       在示例中，我们使用foreach方法打印了每个EdgeRDD的结果。这展示了如何使用EdgeRDD对象操作图的边数据，包括映射、反转和连接等操作。你可以根据自己的需求使用其他EdgeRDD的方法来处理和操作边数据。

       源代码定义了EdgeRDD类及其伴生对象，后者提供了辅助方法。EdgeRDD是对RDD[Edge[ED]]的扩展，用于存储边的列式格式，并可以额外存储边关联的顶点属性。它提供了一系列方法来操作和转换边数据，包括但不限于mapValues、reverse和innerJoin等。EdgeRDD还定义了其他方法，如withTargetStorageLevel、compute和getPartitions等，邦帮源码用于更改目标存储级别、计算和获取分区信息。

       伴生对象中的fromEdges方法用于从一组边创建EdgeRDD，而fromEdgePartitions方法用于从已构造的边分区创建EdgeRDD。这段代码展示了EdgeRDD的主要实现和相关方法，为图计算中的边数据提供了高效的存储和处理能力。

Spark ML系列RandomForestClassifier RandomForestClassificationModel随机森林原理示例源码分析

       Spark ML中的随机森林分类器（RandomForestClassifier）是一个集成学习方法的分类模型。通过使用多个决策树，它进行自助采样与特征随机选择来构建预测模型。其优势在于能够高效处理大量高维数据，对缺失值和噪声具有鲁棒性，并能评估特征重要性，同时训练过程可并行执行提高速度。参数设置如决策树数量、深度和特征选择策略直接影响模型性能和泛化能力，需根据具体问题和数据集调优以获得最佳效果。

       RandomForestClassifier用于Spark ML分类任务，封装在特定类中，支持数据处理与模型训练过程的关键方法。可调整参数优化模型表现，例如特征选择与决策树设置。模型通过构建包含数据转换与训练的Pipeline流程实现自动训练。

       以下为基本示例代码：

        1. 加载数据集并构建特征向量和标签索引。

        2. 将数据集划分为训练集与测试集。

        3. 创建RandomForestClassifier实例，并设定关键参数。

        4. 构建Pipeline并训练模型。

        5. 对测试集进行预测，并评估模型性能，常用指标如多分类准确率。

       代码示例中包含实现RandomForestClassifier类的构造与基本用法，如类成员、常量声明和模型对象定义等。此部分源码用于构造随机森林模型的抽象概念与实现基础。

Spark Core读取ES的分区问题分析

       撰写本文的初衷是因近期一位星球球友面试时，面试官询问了Spark分析ES数据时，生成的RDD分区数与哪些因素相关。

       初步推测，这与分片数有关，但具体关系是什么呢？以下是两种可能的关系：

       1).类似于KafkaRDD的分区与kafka topic分区数的关系，一对一。

       2).ES支持游标查询，那么是否可以对较大的ES索引分片进行拆分，形成多个RDD分区呢？

       下面，我将与大家共同探讨源码，了解具体情况。

       1.Spark Core读取ES

       ES官网提供了elasticsearch-hadoop插件，对于ES 7.x，hadoop和Spark版本的支持如下：

       在此，我使用的ES版本为7.1.1，测试用的Spark版本为2.3.1，没有问题。整合es和spark，导入相关依赖有两种方式：

       a，导入整个elasticsearch-hadoop包

       b，仅导入spark模块的包

       为了方便测试，我在本机启动了一个单节点的ES实例，简单的测试代码如下：

       可以看到，Spark Core读取RDD主要有两种形式的API：

       a，esRDD。这种返回的是一个tuple2类型的RDD，第一个元素是id，第二个是一个map，包含ES的document元素。

       b，esJsonRDD。这种返回的也是一个tuple2类型的RDD，第一个元素依然是id，第二个是json字符串。

       尽管这两种RDD的类型不同，但它们都是ScalaEsRDD类型。

       要分析Spark Core读取ES的并行度，只需分析ScalaEsRDD的getPartitions函数。

       2.源码分析

       首先，导入源码github.com/elastic/elasticsearch-hadoop这个gradle工程，可以直接导入idea，然后切换到7.x版本。

       接下来，找到ScalaEsRDD，发现getPartitions方法是在其父类中实现的，方法内容如下：

       esPartitions是一个lazy型的变量：

       这种声明的原因是什么呢？

       lazy+transient的原因大家可以思考一下。

       RestService.findPartitions方法只是创建客户端获取分片等信息，然后调用，分两种情况调用两个方法：

       a).findSlicePartitions

       这个方法实际上是在5.x及以后的ES版本，同时配置了

       之后，才会执行。实际上就是将ES的分片按照指定大小进行拆分，必然要先进行分片大小统计，然后计算出拆分的分区数，最后生成分区信息。具体代码如下：

       实际上，分片就是通过游标方式，对_doc进行排序，然后按照分片计算得到的分区偏移进行数据读取，组装过程是通过SearchRequestBuilder.assemble方法实现的。

       这个实际上会浪费一定的性能，如果真的要将ES与Spark结合，建议合理设置分片数。

       b).findShardPartitions方法

       这个方法没有疑问，一个RDD分区对应于ES index的一个分片。

       3.总结

       以上就是Spark Core读取ES数据时，分片和RDD分区的对应关系分析。默认情况下，一个ES索引分片对应Spark RDD的一个分区。如果分片数过大，且ES版本在5.x及以上，可以配置参数

       进行拆分。

Spark源码解析2-YarnCluster模式启动

       YARN 模式运行机制主要体现在Yarn Cluster 模式和Yarn Client 模式上。在Yarn Cluster模式下，SparkSubmit、ApplicationMaster 和 CoarseGrainedExecutorBackend 是独立的进程，而Driver 是独立的线程；Executor 和 YarnClusterApplication 是对象。在Yarn Client模式下，SparkSubmit、ApplicationMaster 和 YarnCoarseGrainedExecutorBackend 也是独立的进程，而Executor和Driver是对象。

       在源码中，SparkSubmit阶段首先执行Spark提交命令，底层执行的是开启SparkSubmit进程的命令。代码中，SparkSubmit从main()开始，根据运行模式获取后续要反射调用的类名赋给元组中的ChildMainClass。如果是Yarn Cluster模式，则为YarnClusterApplication；如果是Yarn Client模式，则为主类用户自定义的类。接下来，获取ChildMainClass后，通过反射调用main方法的过程，反射获取类然后通过构造器获取一个示例并多态为SparkApplication，再调用它的start方法。随后调用YarnClusterApplication的start方法。在YarnClient中，new一个Client对象，其中包含了yarnClient = YarnClient.createYarnClient属性，这是Yarn在SparkSubmit中的客户端，yarnClient在第行初始化和开始，即连接Yarn集群或RM。之后就可以通过这个客户端与Yarn的RM进行通信和提交应用，即调用run方法。

       ApplicationMaster阶段主要涉及开启一个Driver新线程、AM向RM注册、AM向RM申请资源并处理、封装ExecutorBackend启动命令以及AM向NM通信提交命令由NM启动ExecutorBackend。在ApplicationMaster进程中，首先开启Driver线程，开始运行用户自定义代码，创建Spark程序入口SparkContext，接着创建RDD，生成job，划分阶段提交Task等操作。

       在申请资源之前，AM主线程创建了Driver的终端引用，作为参数传入createAllocator()，因为Executor启动后需要向Driver反向注册，所以启动过程必须封装Driver的EndpointRef。AM主线程向RM申请获取可用资源Container，并处理这些资源。ExecutorBackend阶段尚未完成，后续内容待补充。