rdd的特点

       rdd的特点如下：

       1、RDD是Spark提供的核心抽象，全称为ResillientDistributedDataset，即弹性分布式数据集。

       2、RDD在抽象上来说是一种元素集合，包含了数据。它是被分区的，分为多个分区，每个分区分布在集群中的不同节点上，从而让RDD中的数据可以被并行操作。

       3、RDD通常通过Hadoop上的文件，即HDFS文件或者Hive表，来进行创建；有时也可以通过应用程序中的集合来创建。

       4、RDD最重要的特性就是，提供了容错性，可以自动从节点失败中恢复过来。即如果某个节点上的RDDpartition，因为节点故障，导致数据丢了，那么RDD会自动通过自己的数据来源重新计算该partition。这一切对使用者是透明的。

       5、RDD的数据默认情况下存放在内存中的，但是在内存资源不足时，Spark会自动将RDD数据写入磁盘。

Apache 两个开源项目比较：Flink vs Spark

       时间久远，我对云计算与大数据已感生疏，分析尤其是源码Flink的崛起。自动驾驶平台需云计算支撑，分析包括机器学习、源码深度学习训练、分析弹珠台源码高清地图、源码模拟仿真模块，分析以及车联网。源码近日看到一篇Spark与Flink的分析比较文章，遂转发分享，源码以便日后重新学习该领域新知识。分析

       Apache Flink作为新一代通用大数据处理引擎，源码致力于整合各类数据负载。分析它似乎与Apache Spark有着相似目标。源码两者都旨在构建一个单一平台，用于批处理、流媒体、交互式、图形处理、机器学习等。因此，Flink与Spark在理念上并无太大差异。但在实施细节上，它们却存在显著区别。

       以下比较Spark与Flink的不同之处。尽管两者在某些方面存在相似之处，但也有许多不同之处。zblog单页源码

       1. 抽象

       在Spark中，批处理采用RDD抽象，而流式传输使用DStream。Flink为批处理数据集提供数据集抽象，为流应用程序提供DataStream。尽管它们听起来与RDD和DStreams相似，但实际上并非如此。

       以下是差异点：

       在Spark中，RDD在运行时表示为Java对象。随着project Tungsten的推出，它略有变化。但在Apache Flink中，数据集被表示为一个逻辑计划。这与Spark中的Dataframe相似，因此在Flink中可以像使用优化器优化的一等公民那样使用API。然而，Spark RDD之间并不进行任何优化。

       Flink的数据集类似Spark的Dataframe API，在执行前进行了优化。

       在Spark 1.6中，数据集API被添加到spark中，可能最终取代RDD抽象。

       在Spark中，所有不同的抽象，如DStream、Dataframe都建立在RDD抽象之上。但在Flink中，安卓源码 romDataset和DataStream是基于顶级通用引擎构建的两个独立抽象。尽管它们模仿了类似的API，但在DStream和RDD的情况下，无法将它们组合在一起。尽管在这方面有一些努力，但最终结果还不够明确。

       无法将DataSet和DataStream组合在一起，如RDD和DStreams。

       因此，尽管Flink和Spark都有类似的抽象，但它们的实现方式不同。

       2. 内存管理

       直到Spark 1.5，Spark使用Java堆来缓存数据。虽然项目开始时更容易，但它导致了内存不足（OOM）问题和垃圾收集（gc）暂停。因此，从1.5开始，Spark进入定制内存管理，称为project tungsten。

       Flink从第一天起就开始定制内存管理。实际上，这是Spark向这个方向发展的灵感之一。不仅Flink将数据存储在它的自定义二进制布局中，它确实直接对二进制数据进行操作。在Spark中，所有数据帧操作都直接在Spark 1.5的project tungsten二进制数据上运行。

       在JVM上执行自定义内存管理可以提高性能并提高资源利用率。串口波形显示源码

       3. 实施语言

       Spark在Scala中实现。它提供其他语言的API，如Java、Python和R。

       Flink是用Java实现的。它确实提供了Scala API。

       因此，与Flink相比，Spark中的选择语言更好。在Flink的一些scala API中，java抽象也是API的。这会有所改进，因为已经使scala API获得了更多用户。

       4. API

       Spark和Flink都模仿scala集合API。所以从表面来看，两者的API看起来非常相似。

       5. 流

       Apache Spark将流式处理视为快速批处理。Apache Flink将批处理视为流处理的特殊情况。这两种方法都具有令人着迷的含义。

       以下是两种不同方法的差异或含义：

       Apache Flink提供事件级处理，也称为实时流。它与Storm模型非常相似。

       Spark只有不提供事件级粒度的最小批处理（mini-batch）。这种方法被称为近实时。

       Spark流式处理是更快的批处理，Flink批处理是有限的流处理。

       虽然大多数应用程序都可以近乎实时地使用，列表页面 div 源码但很少有应用程序需要事件级实时处理。这些应用程序通常是Storm流而不是Spark流。对于他们来说，Flink将成为一个非常有趣的选择。

       运行流处理作为更快批处理的优点之一是，我们可以在两种情况下使用相同的抽象。Spark非常支持组合批处理和流数据，因为它们都使用RDD抽象。

       在Flink的情况下，批处理和流式传输不共享相同的API抽象。因此，尽管有一些方法可以将基于历史文件的数据与流相结合，但它并不像Spark那样干净。

       在许多应用中，这种能力非常重要。在这些应用程序中，Spark代替Flink流式传输。

       由于最小批处理的性质，Spark现在对窗口的支持非常有限。允许根据处理时间窗口批量处理。

       与其他任何系统相比，Flink提供了非常灵活的窗口系统。Window是Flink流API的主要焦点之一。它允许基于处理时间、数据时间和无记录等的窗口。这种灵活性使Flink流API与Spark相比非常强大。

       6. SQL界面

       截至目前，最活跃的Spark库之一是spark-sql。Spark提供了像Hive一样的查询语言和像DSL这样的Dataframe来查询结构化数据。它是成熟的API并且在批处理中广泛使用，并且很快将在流媒体世界中使用。

       截至目前，Flink Table API仅支持DSL等数据帧，并且仍处于测试阶段。有计划添加sql接口，但不确定何时会落在框架中。

       目前为止，Spark与Flink相比有着不错的SQL故事。

       7. 数据源集成

       Spark数据源API是框架中最好的API之一。数据源API使得所有智能资源如NoSQL数据库、镶嵌木地板、优化行列（Optimized Row Columnar，ORC）成为Spark上的头等公民。此API还提供了在源级执行谓词下推（predicate push down）等高级操作的功能。

       Flink仍然在很大程度上依赖于map / reduce InputFormat来进行数据源集成。虽然它是足够好的提取数据API，但它不能巧妙地利用源能力。因此Flink目前落后于目前的数据源集成技术。

       8. 迭代处理

       Spark最受关注的功能之一就是能够有效地进行机器学习。在内存缓存和其他实现细节中，它是实现机器学习算法的真正强大的平台。

       虽然ML算法是循环数据流，但它表示为Spark内部的直接非循环图。通常，没有分布式处理系统鼓励循环数据流，因为它们变得难以理解。

       但是Flink对其他人采取了一些不同的方法。它们在运行时支持受控循环依赖图（cyclic dependence graph）。这使得它们与DAG表示相比以非常有效的方式表示ML算法。因此，Flink支持本机平台中的迭代，与DAG方法相比，可实现卓越的可扩展性和性能。

       9. 流作为平台与批处理作为平台

       Apache Spark来自Map / Reduce时代，它将整个计算表示为数据作为文件集合的移动。这些文件可能作为磁盘上的阵列或物理文件驻留在内存中。这具有非常好的属性，如容错等。

       但是Flink是一种新型系统，它将整个计算表示为流处理，其中数据有争议地移动而没有任何障碍。这个想法与像akka-streams这样的新的反应流系统非常相似。

       . 成熟

       Flink像批处理这样的部分已经投入生产，但其他部分如流媒体、Table API仍在不断发展。这并不是说在生产中就没人使用Flink流。

请问各位大神,spark的ml和mllib两个包区别和联系?!?

       在技术角度上，Spark的ML和Mllib包处理数据集的方式不同。ML包面向的是Dataset，具体来说是Dataframe，而Mllib则直接面对RDD。Dataset和RDD之间的区别在于，Dataset是在RDD基础上进行深度优化的版本。

       Dataset优化了性能和静态类型分析，提供了类似于SQL语言的功能，能够在编译时捕获错误。相比于RDD，Dataset的combinators（如map和foreach等）性能表现更优。

       在编程过程中，构建机器学习算法的方式也有所不同。ML包提倡使用pipelines进行数据处理。想象数据如同水流，从管道的一端流入，另一端流出。具体实现为：DataFrame --> Pipeline --> 新DataFrame。Pipeline是通过连接Transformer和Estimator实现的数据处理流程。

       Transformer的输入是DataFrame，输出同样是DataFrame。而Estimator的输入是DataFrame，输出则是一个Transformer。这种流程使得数据处理逻辑清晰，易于理解和维护。

SPARK- - Spark支持unpivot源码分析

       unpivot是数据库系统中用于列转行的内置函数，如SQL SERVER, Oracle等。以数据集tb1为例，每个数字代表某个人在某个学科的成绩。若要将此表扩展为三元组，可使用union实现。但随列数增加，SQL语句变长。许多SQL引擎提供内置函数unpivot简化此过程。unpivot使用时需指定保留列、进行转行的列、新列名及值列名。

       SPARK从SPARK-版本开始支持DataSet的unpivot函数，逐步扩展至pyspark与SQL。在Dataset API中，ids为要保留的Column数组，Column类提供了从String构造Column的隐式转换，方便使用。利用此API，可通过unpivot函数将数据集转换为所需的三元组。values表示转行列，variableColumnName为新列名，valueColumnName为值列名。

       Analyser阶段解析unpivot算子，将逻辑执行计划转化为物理执行计划。当用户开启hive catalog，SPARK SQL根据表名和metastore URL查找表元数据，转化为Hive相关逻辑执行计划。物理执行计划如BroadcastHashJoinExec，表示具体的执行策略。规则ResolveUnpivot将包含unpivot的算子转换为Expand算子，在物理执行计划阶段执行。此转换由开发者自定义规则完成，通过遍历逻辑执行计划树，根据节点类型及状态进行不同处理。

       unpivot函数实现过程中，首先将原始数据集投影为包含ids、variableColumnName、valueColumnName的列，实现语义转换。随后，通过map函数处理values列，构建新的行数据，最终返回Expand算子。在物理执行计划阶段，Expand算子将数据转换为所需形式，实现unpivot功能。

       综上所述，SPARK内置函数unpivot的实现通过解析列参数，组装Expand算子完成，为用户提供简便的列转行功能。通过理解此过程，可深入掌握SPARK SQL的开发原理与内在机制。

RDD，DataFrame和DataSet的区别

       RDD、DataFrame和DataSet是容易产生混淆的概念，必须对其相互之间对比，才可以知道其中异同。

       RDD和DataFrame

       RDD-DataFrame

       ä¸Šå›¾ç›´è§‚地体现了DataFrame和RDD的区别。左侧的RDD[Person]虽然以Person为类型参数，但Spark框架本身不了解

       Person类的内部结构。而右侧的DataFrame却提供了详细的结构信息，使得Spark

       SQL可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么。DataFrame多了数据的结构信息，即schema。RDD是分布式的

       Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外，更重要的特点是提升执行效

       çŽ‡ã€å‡å°‘数据读取以及执行计划的优化，比如filter下推、裁剪等。

       æå‡æ‰§è¡Œæ•ˆçŽ‡

       RDD

       API是函数式的，强调不变性，在大部分场景下倾向于创建新对象而不是修改老对象。这一特点虽然带来了干净整洁的API，却也使得Spark应用程序在运

       è¡ŒæœŸå€¾å‘于创建大量临时对象，对GC造成压力。在现有RDD

       API的基础之上，我们固然可以利用mapPartitions方法来重载RDD单个分片内的数据创建方式，用复用可变对象的方式来减小对象分配和GC的

       å¼€é”€ï¼Œä½†è¿™ç‰ºç‰²äº†ä»£ç çš„可读性，而且要求开发者对Spark运行时机制有一定的了解，门槛较高。另一方面，Spark

       SQL在框架内部已经在各种可能的情况下尽量重用对象，这样做虽然在内部会打破了不变性，但在将数据返回给用户时，还会重新转为不可变数据。利用

       DataFrame API进行开发，可以免费地享受到这些优化效果。

       å‡å°‘数据读取

       åˆ†æžå¤§æ•°æ®ï¼Œæœ€å¿«çš„方法就是 ——忽略它。这里的“忽略”并不是熟视无睹，而是根据查询条件进行恰当的剪枝。

       ä¸Šæ–‡è®¨è®ºåˆ†åŒºè¡¨æ—¶æåˆ°çš„分区剪 枝便是其中一种——当查询的过滤条件中涉及到分区列时，我们可以根据查询条件剪掉肯定不包含目标数据的分区目录，从而减少IO。

       å¯¹äºŽä¸€äº›â€œæ™ºèƒ½â€æ•°æ®æ ¼ 式，Spark

       SQL还可以根据数据文件中附带的统计信息来进行剪枝。简单来说，在这类数据格式中，数据是分段保存的，每段数据都带有最大值、最小值、null值数量等

       ä¸€äº›åŸºæœ¬çš„统计信息。当统计信息表名某一数据段肯定不包括符合查询条件的目标数据时，该数据段就可以直接跳过(例如某整数列a某段的最大值为，而查

       è¯¢æ¡ä»¶è¦æ±‚a > )。

       æ­¤å¤–，Spark SQL也可以充分利用RCFile、ORC、Parquet等列式存储格式的优势，仅扫描查询真正涉及的列，忽略其余列的数据。

       æ‰§è¡Œä¼˜åŒ–

       äººå£æ•°æ®åˆ†æžç¤ºä¾‹

       ä¸ºäº†è¯´æ˜ŽæŸ¥è¯¢ä¼˜åŒ–，我们来看上图展示的人口数据分析的示例。图中构造了两个DataFrame，将它们join之后又做了一次filter操作。如

       æžœåŽŸå°ä¸åŠ¨åœ°æ‰§è¡Œè¿™ä¸ªæ‰§è¡Œè®¡åˆ’，最终的执行效率是不高的。因为join是一个代价较大的操作，也可能会产生一个较大的数据集。如果我们能将filter

        下推到 join下方，先对DataFrame进行过滤，再join过滤后的较小的结果集，便可以有效缩短执行时间。而Spark

       SQL的查询优化器正是这样做的。简而言之，逻辑查询计划优化就是一个利用基于关系代数的等价变换，将高成本的操作替换为低成本操作的过程。

       å¾—到的优化执行计划在转换成物 理执行计划的过程中，还可以根据具体的数据源的特性将过滤条件下推至数据源内。最右侧的物理执行计划中Filter之所以消失不见，就是因为溶入了用于执行最终的读取操作的表扫描节点内。

       å¯¹äºŽæ™®é€šå¼€å‘者而言，查询优化 器的意义在于，即便是经验并不丰富的程序员写出的次优的查询，也可以被尽量转换为高效的形式予以执行。

       RDD和DataSet

       DataSet以Catalyst逻辑执行计划表示，并且数据以编码的二进制形式被存储，不需要反序列化就可以执行sorting、shuffle等操作。

       DataSet创立需要一个显式的Encoder，把对象序列化为二进制，可以把对象的scheme映射为SparkSQl类型，然而RDD依赖于运行时反射机制。

       é€šè¿‡ä¸Šé¢ä¸¤ç‚¹ï¼ŒDataSet的性能比RDD的要好很多。

       DataFrame和DataSet

       Dataset可以认为是DataFrame的一个特例，主要区别是Dataset每一个record存储的是一个强类型值而不是一个Row。因此具有如下三个特点：

       DataSet可以在编译时检查类型

       å¹¶ä¸”是面向对象的编程接口。用wordcount举例：

       //DataFrame

       // Load a text file and interpret each line as a java.lang.String

       val ds = sqlContext.read.text("/home/spark/1.6/lines").as[String]

       val result = ds

        .flatMap(_.split(" ")) // Split on whitespace

        .filter(_ != "") // Filter empty words

        .toDF() // Convert to DataFrame to perform aggregation / sorting

        .groupBy($"value") // Count number of occurences of each word

        .agg(count("*") as "numOccurances")

        .orderBy($"numOccurances" desc) // Show most common words first

       åŽé¢ç‰ˆæœ¬DataFrame会继承DataSet，DataFrame是面向Spark SQL的接口。

       //DataSet,完全使用scala编程，不要切换到DataFrame

       val wordCount =

        ds.flatMap(_.split(" "))

        .filter(_ != "")

        .groupBy(_.toLowerCase()) // Instead of grouping on a column expression (i.e. $"value") we pass a lambda function

        .count()

       DataFrame和DataSet可以相互转化， df.as[ElementType] 这样可以把DataFrame转化为DataSet， ds.toDF() 这样可以把DataSet转化为DataFrame。

谈谈RDD，DataFrame，Dataset的区别和各自的优势

       RDD、DataFrame、Dataset全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利

        2、三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算，极端情况下，如果代码里面有创建、转换，但是后面没有在Action中使用对应的结果，在执行时会被直接跳过，如