皮皮网

1.Apache 两个开源项目比较：Flink vs Spark
2.SPARK-38864 - Spark支持unpivot源码分析
3.谈谈RDD，DataFrame，Dataset的区别和各自的优势
4.Spark RDD，DataFrame和DataSet的区别
5.请问各位大神,源码spark的ml和mllib两个包区别和联系?!?
6.从Spark2.4升级到3.0哪些变化

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

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

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

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

       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中，Dataset和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批处理是有限的流处理。

       虽然大多数应用程序都可以近乎实时地使用，但很少有应用程序需要事件级实时处理。这些应用程序通常是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- - 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:

       1、RDD一般和spark mlib同时使用

       2、RDD不支持sparksql操作

       DataFrame:

       ä¸ŽRDD和Dataset不同，DataFrame每一行的类型固定为Row，只有通过解析才能获取各个字段的值

Spark RDD，DataFrame和DataSet的区别

       RDD：A Resilient Distributed Dataset (RDD), the basic abstraction in Spark.

       rdd是一个分布式的数据集，数据分散在分布式集群的各台机器上

       dataframe更像是一张关系型数据表，是一种spark独有的数据格式吧，这种格式的数据可以使用sqlcontext里面的函数

请问各位大神,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。这种流程使得数据处理逻辑清晰，易于理解和维护。

从Spark2.4升级到3.0哪些变化

       在升级到Spark 3.0后，Dataset和DataFrame API的unionAll不再被弃用，它是union的别名。在Spark 2.4及以下版本中，如果键的类型为非结构化类型（例如int、string、array等），则Dataset.groupByKey的结果会导致带有错误名称"value"的分组数据集，这种行为令人困惑，并使聚合查询的模式出乎意料。从Spark 3.0开始，分组属性被命名为"key"。通过新添加的配置spark.sql.legacy.dataset.nameNonStructGroupingKeyAsValue，可以保留旧的行为，默认值为false。

       在Spark 3.0中，列元数据将始终在API Column.name 和 Column.as 中传播。在Spark 2.4及更早版本中，NamedExpression 的元数据在调用API时作为新列的explicitMetadata设置，即使底层的NamedExpression 更改了元数据，也不会更改。要恢复Spark 2.4之前的行为，可以使用带有显式元数据的API as(alias: String, metadata: Metadata)。

       某些不合理的类型转换，在Spark 3.0中被禁止。例如，将string转换为int或double转换为boolean是不允许的。如果值超出列的数据类型范围，则会抛出运行时异常。在Spark 2.4及以下版本中，只要是有效的Cast，在插入表时允许进行类型转换。当向整数字段插入超出范围的值时，插入的是值的低位（与Java/Scala数值类型转换相同）。此行为由选项spark.sql.storeAssignmentPolicy控制，默认值为"ANSI"。将该选项设置为"Legacy"可以恢复之前的行为。

       在Spark 2.4及以下版本中，即使指定的键是SparkConf条目的键，并且没有生效，SET命令也可以正常工作，而不会发出任何警告，因为该命令不会更新SparkConf，但这种行为可能会让用户感到困惑。在3.0中，如果使用了SparkConf键，该命令将失败。可以通过将spark.sql.legacy.setCommandRejectsSparkCoreConfs设置为false来禁用此检查。

       在Spark 2.4及以下版本中，在取消缓存操作之前并没有保留缓存名称和存储级别。因此，缓存名称和存储级别可能会意外更改。在Spark 3.0中，首先保留缓存名称和存储级别以进行缓存重建。这有助于在刷新表时保持一致的缓存行为。

       在Spark 3.0中，以下属性列出的命令被保留；如果在CREATE DATABASE ... WITH DBPROPERTIES和ALTER TABLE ... SET TBLPROPERTIES等位置指定了保留的属性，命令将失败。您需要使用它们特定的子句来指定它们，例如CREATE DATABASE test COMMENT 'any comment' LOCATION 'some path'。您可以将 spark.sql.legacy.notReserveProperties 设置为 true，以忽略 ParseException，在这种情况下，这些属性将被静默删除，例如：SET DBPROPERTIES('location'='/tmp') 将不起作用。在Spark 2.4及以下版本中，这些属性既不是保留属性，也没有副作用，例如：SET DBPROPERTIES('location'='/tmp') 不会更改数据库的位置，而只会创建一个无头属性，就像 'a'='b' 一样。

       Spark 2.4及以下版本：在取消缓存操作之前并没有保留缓存名称和存储级别。因此，缓存名称和存储级别可能会意外更改。在Spark 3.0中，首先保留缓存名称和存储级别以进行缓存重建。这有助于在刷新表时保持一致的缓存行为。

       在Spark 3.0中，查询引擎变化会影响CSV/JSON数据源以及使用用户指定模式进行解析和格式化的unix_timestamp、date_format、to_unix_timestamp、from_unixtime、to_date、to_timestamp函数。在Spark 3.0中，我们在sql-ref-datetime-pattern.md中定义了自己的模式字符串，它是通过底层的java.time.format.DateTimeFormatter实现的。新的实现对其输入进行严格检查。例如，如果模式为yyyy-MM-dd，则无法解析-- ::时间戳，因为解析器没有消耗整个输入。另一个例子是，使用dd/MM/yyyy hh:mm模式无法解析// :输入，因为hh表示的是范围为1-的小时。而在Spark 2.4及以下版本中，使用java.text.SimpleDateFormat进行时间戳/日期字符串转换，并且支持的模式在 simpleDateFormat中描述。可以通过将spark.sql.legacy.timeParserPolicy设置为LEGACY来恢复旧的行为。

       在Spark 3.0中，weekofyear、weekday、dayofweek、date_trunc、from_utc_timestamp、to_utc_timestamp和unix_timestamp函数使用java.time API来计算一年中的周数、一周中的天数，以及在UTC时区之间进行TimestampType值的转换。JDBC选项lowerBound和upperBound与将字符串转换为TimestampType/DateType值相同。该转换基于Proleptic Gregorian日历，并由SQL配置spark.sql.session.timeZone定义的时区确定。而在Spark 2.4及以下版本中，该转换基于混合日历（朱利安+格里高利）和默认系统时区。

       从字符串创建带有类型的TIMESTAMP和DATE字面量。在Spark 3.0中，将字符串转换为带有类型的TIMESTAMP/DATE字面量是通过将其强制转换为TIMESTAMP/DATE值来执行的。例如，TIMESTAMP '-- ::'在语义上等于CAST('-- ::' AS TIMESTAMP)。当输入字符串不包含时区信息时，使用SQL配置spark.sql.session.timeZone中定义的时区。而在Spark 2.4及以下版本中，该转换基于JVM系统时区。默认时区的不同来源可能会改变带有类型的TIMESTAMP和DATE字面量的行为。

       例如，SELECT date 'tomorrow' - date 'yesterday'; 应该输出 2。以下是特殊的时间戳值：例如，SELECT timestamp 'tomorrow';。

       自Spark 3.0起，使用EXTRACT表达式从日期/时间戳值中提取秒字段时，结果将是一个带有两位秒部分和六位小数部分的DecimalType(8, 6)值，精确到微秒。例如，extract(second from to_timestamp('-- ::.1'))的结果是.。而在Spark 2.4及更早版本中，它返回一个IntegerType值，对于前面的示例，结果为。

       在升级到Spark 3.0后，内置Hive从1.2升级到2.3，这带来了以下影响：您可能需要根据要连接的Hive元存储的版本设置spark.sql.hive.metastore.version和spark.sql.hive.metastore.jars。例如：如果您的Hive元存储版本是1.2.1，则将spark.sql.hive.metastore.version设置为1.2.1，spark.sql.hive.metastore.jars设置为maven。您需要将自定义的SerDes迁移到Hive 2.3或使用带有hive-1.2配置文件构建自己的Spark。有关更多详细信息，请参见 HIVE-。当使用SQL进行脚本转换时，Hive 1.2和Hive 2.3之间的字符串表示可能不同，具体取决于hive的行为。在Hive 1.2中，字符串表示省略尾随零。但在Hive 2.3中，如果需要，它总是填充为位数字，并包含尾随零。