设为首页 加入收藏

TOP

Spark2.0较于Spark1.x的更新之处
2019-03-15 01:05:20 】 浏览:11
Tags:Spark2.0 较于 Spark1.x 更新
版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/Jorocco/article/details/85683340

Spark Core&Spark SQL API

1、dataframe与dataset统一,dataframe只是dataset[Row]的类型别名,统一了Dataframe和Dataset两套API。

从Spark 2.0开始,Dataframe就只是Dataset[Row]的一个别名,不再是一个单独的类了。无论是typed方法(map、filter、groupByKey等)还是untyped方法(select、groupBy等),都通过Dataset来提供。而且Dataset API将成为Spark的新一代流式计算框架——structured streaming的底层计算引擎。但是由于Python和R这两个语言都不具备compile-time type-safety的特性,所以就没有引入Dataset API,所以这两种语言中的主要编程接口还是Dataframe。

2、SparkSession:统一SQLContext和HiveContext,新的上下文入口

SparkSession是新的Spark上下文以及入口,用于合并SQLContext和HiveContext,并替代它们。因为以前提供了SQLContext和HiveContext两种上下文入口,因此用户有时会有些迷惑,到底该使用哪个接口。现在好了,只需要使用一个统一的SparkSession即可。但是为了向后兼容性,SQLContext和HiveContext还是保留下来了。

3、为SparkSession开发的一种新的流式调用的configuration api
4、accumulator功能增强:便捷api、web ui支持、性能更高
5、dataset的增强聚合api

Spark Core&Spark SQL SQL

1、支持sql 2003标准

在标准化SQL支持方面,引入了新的ANSI-SQL解析器,提供标准化SQL的解析功能,而且还提供了子查询的支持。Spark现在可以运行完整的99个TPC-DS查询,这就要求Spark包含大多数SQL 2003标准的特性。这么做的好处在于,SQL一直是大数据应用领域的一个最广泛接受的标准,比如说Hadoop,做大数据的企业90%的时间都在用Hive,写SQL做各种大数据的统计和分析。因此Spark SQL提升对SQL的支持,可以大幅度减少用户将应用从其他技术(比如Oracle、Hive等)迁移过来的成本。

2、支持ansi-sql和hive ql的sql parser
3、支持ddl命令
4、支持子查询:in/not in、exists/not exists

Spark Core&Spark SQL new feature

1、支持csv文件
2、支持缓存和程序运行的堆外内存管理
3、支持hive风格的bucket表
4、 支持近似概要统计,包括近似分位数、布隆过滤器、最小略图

Spark Core&Spark SQL性能

1、通过whole-stage code generation技术将spark sql和dataset的性能提升2~10倍

Spark 2.0的一个重大的特点就是搭载了最新的第二代tungsten引擎。CPU的大部分时间都浪费在了一些无用的工作上,比如说virtual function call,或者从CPU缓冲区中读写数据。第二代tungsten引擎吸取了现代编译器以及并行数据库的一些重要的思想,并且应用在了spark的运行机制中。其中一个核心的思想,就是在运行时动态地生成代码,在这些自动动态生成的代码中,可以将所有的操作都打包到一个函数中,这样就可以避免多次virtual function call,而且还可以通过cpu register来读写中间数据,而不是通过cpu cache来读写数据。上述技术整体被称作“whole-stage code generation”,中文也可以叫“全流程代码生成”。

深入剖析Spark 2.x的第二代tungsten引擎原理之前,先看一下当前的Spark的工作原理。我们可以通过一个SQL来举例,这个SQL扫描了单个表,然后对属性等于指定值的记录进行汇总计数。SQL语句如下:select count(*) from store_sales where ss_item_sk=1000。

要执行这个查询,Spark 1.x会使用一种最流行、最经典的查询求值策略,该策略主要基于Volcano Iterator Model。在这种模型中,一个查询会包含多个operator,每个operator都会实现一个接口,提供一个next()方法,该方法返回operator tree中的下一个operator。
在这里插入图片描述

loop unrolling解释(小白的方式)
for(int i = 0; i < 10; i++) { System.out.println(i) }
System.out.println(1)
System.out.println(2)
System.out.println(3)
......

手写代码的好处就在于,它是专门为实现这个功能而编写的,代码简单,因此可以吸收上述所有优点,包括避免虚函数调用,将中间数据保存在CPU寄存器中,而且还可以被底层硬件进行for循环的自动优化。

手写的代码的性能比Volcano Iterator Model高了一整个数量级,而这其中的原因包含以下几点:

1、避免了virtual function dispatch:在Volcano Iterator Model中,至少需要调用一次next()函数来获取下一个operator。这些函数调用在操作系统层面,会被编译为virtual function dispatch。而手写代码中,没有任何的函数调用逻辑。虽然说,现代的编译器已经对虚函数调用进行了大量的优化,但是该操作还是会执行多个CPU指令,并且执行速度较慢,尤其是当需要成百上千次地执行虚函数调用时。

2、通过CPU Register存取中间数据,而不是内存缓冲:在Volcano Iterator Model中,每次一个operator将数据交给下一个operator,都需要将数据写入内存缓冲中。然而在手写代码中,JVM JIT编译器会将这些数据写入CPU Register。CPU从内存缓冲种读写数据的性能比直接从CPU Register中读写数据,要低了一个数量级。

3、Loop Unrolling和SIMD:现代的编译器和CPU在编译和执行简单的for循环时,性能非常地高。编译器通常可以自动对for循环进行unrolling,并且还会生成SIMD指令以在每次CPU指令执行时处理多条数据。CPU也包含一些特性,比如pipelining,prefetching,指令reordering,可以让for循环的执行性能更高。然而这些优化特性都无法在复杂的函数调用场景中施展,比如Volcano Iterator Model。

上面讲解了手工编写的代码的性能,为什么比Volcano Iterator Model要好。所以如果要对Spark进行性能优化,一个思路就是在运行时动态生成代码,以避免使用Volcano模型,转而使用性能更高的代码方式。要实现上述目的,就引出了Spark第二代Tungsten引擎的新技术,whole-stage code generation。通过该技术,SQL语句编译后的operator-treee中,每个operator执行时就不是自己来执行逻辑了,而是通过whole-stage code generation技术,动态生成代码,生成的代码中会尽量将所有的操作打包到一个函数中,然后再执行动态生成的代码。

举例来说:
就以上一讲的SQL语句来作为示例,Spark会自动生成以下代码。如果只是一个简单的查询,那么Spark会尽可能就生成一个stage,并且将所有操作打包到一起。但是如果是复杂的操作,就可能会生成多个stage。
在这里插入图片描述
Spark提供了explain()方法来查看一个SQL的执行计划,而且这里面是可以看到通过whole-stage code generation生成的代码的执行计划的。如果看到一个步骤前面有个*符号,那么就代表这个步骤是通过该技术自动生成的。在这个例子中,Range、Filter和Aggregation都是自动生成的,Exchange不是自动生成的,因为这是一个网络传输数据的过程。
在这里插入图片描述
从Spark 1.1版本开始,就一直听说有code generation类的feature引入,这跟spark 2.0中的这个技术有什么不同呢。实际上在spark 1.x版本中,code generation技术仅仅被使用在了expression evoluation方面(比如a + 1),即表达式求值,还有极其少数几个算子上(比如filter等)。而spark 2.0中的whole-stage code generation技术是应用在整个spark运行流程上的。

2、通过vectorization技术提升parquet文件的扫描吞吐量

通过vectarization技术将parquet文件扫描的吞吐量提升了3倍以上。

对于很多查询操作,whole-stage code generation技术都可以很好地优化其性能。但是有一些特殊的操作,却无法很好的使用该技术,比如说比较复杂一些操作,如parquet文件扫描、csv文件解析等,或者是跟其他第三方技术进行整合。

如果要在上述场景提升性能,spark引入了另外一种技术,称作“vectorization”,即向量化。向量化的意思就是避免每次仅仅处理一条数据,相反,将多条数据通过面向列的方式来组织成一个一个的batch,然后对一个batch中的数据来迭代处理。每次next()函数调用都返回一个batch的数据,这样可以减少virtual function dispatch的开销。同时通过循环的方式来处理,也可以使用编译器和CPU的loop unrolling等优化特性。
在这里插入图片描述
这种向量化的技术,可以使用到之前说的3个点中的2个点。即,减少virtual function dispatch,以及进行loop unrolling优化。但是还是需要通过内存缓冲来读写中间数据的。所以,仅仅当实在无法使用whole-stage code generation时,才会使用vectorization技术。有人做了一个parquet文件读取的实验,采用普通方式以及向量化方式,性能也能够达到一个数量级的提升。

上述的whole-stage code generation技术,能否保证将spark 2.x的性能比spark 1.x来说提升10倍以上呢?这是无法完全保证的。虽然说目前的spark架构已经搭载了目前世界上最先进的性能优化技术,但是并不是所有的操作都可以大幅度提升性能的。简单来说,CPU密集型的操作,可以通过这些新技术得到性能的大幅度提升,但是很多IO密集型的操作,比如shuffle过程的读写磁盘,是无法通过该技术提升性能的。在未来,spark会花费更多的精力在优化IO密集型的操作的性能上。

3、提升orc文件的读写性能
4、提升catalyst查询优化器的性能

spark 2.0中,除了whole-stage code generation技术以外,还使用了其他一些新技术来提升性能。比如说对Spark SQL的catalyst查询优化器做了一些性能优化,来提升对一些常见查询的优化效率,比如null值处理等。

5、通过native实现方式提升窗口函数的性能
6、对某些数据源进行自动文件合并

Spark MLlib

1、 spark mllib未来将主要基于dataset api来实现,基于rdd的api转为维护阶段

Spark 2.0中,spark.ml包下的机器学习API,主要是基于Dataframe/Dataset来实现的,未来将会成为主要发展的API接口。原先老的基于RDD的spark.mllib包的机器学习API还会保留着,为了向后兼容性,但是未来主要会基于spark.ml包下的接口来进行开发。而且用户使用基于Dataframe/Dataset的新API,还能够对算法模型和pipeline进行持久化保存以及加载。

2、基于dataframe的api,支持持久化保存和加载模型和pipeline
3、基于dataframe的api,支持更多算法,包括二分kmeans、高斯混合、maxabsscaler等
4、spark R支持mllib算法,包括线性回归、朴素贝叶斯、kmeans、多元回归等

Spark 2.0中,为SparkR提供了分布式的机器学习算法,包括经典的Generalized Linear Model,朴素贝叶斯,Survival Regression,K-means等。此外SparkR还支持用户自定义的函数,即UDF。

5、pyspark支持更多mllib算法,包括LDA、高斯混合、泛化线性回顾等
6、基于dataframe的api,向量和矩阵使用性能更高的序列化机制

Spark Streaming

1、发布测试版的structured streaming。

基于spark sql和catalyst引擎构建
支持使用dataframe风格的api进行流式计算操作
catalyst引擎能够对执行计划进行优化
Structured Streaming提供了与批处理计算类似的API。要开发一个流式计算应用,开发人员只要使用Dataframe/Dataset API编写与批处理计算一样的代码即可,Structured Streaming会自动将这些类似批处理的计算代码增量式地应用到持续不断进入的新数据上。这样,开发人员就不需要花太多时间考虑状态管理、容错、与离线计算的同步等问题。Structured Streaming可以保证,针对相同的数据,始终与离线计算产出完全一样的计算结果。

Structured Streaming还提供了与存储系统的事务整合。它会进行自动的容错管理以及数据一致性的管理,如果开发人员要写一个应用程序来更新数据库,进而提供一些实时数据服务,与静态数据进行join,或者是在多个存储系统之间移动数据,那么Structured Streaming可以让这些事情更加简单。

Structured Streaming与Spark其余的组件都能够进行完美的整合。比如可以通过Spark SQL对实时数据进行统计分析,与静态数据进行join,还有其他的使用dataframe/dataset的组件,这样就可以让开发人员构建完整的流式计算引用,而不仅仅只是一个流式计算引擎而已。在未来,Spark会将Structured Streaming与Spark MLlib的整合做的更好。

2、基于dstream的api支持kafka 0.10版本

依赖管理、打包和操作

1、不再需要在生产环境部署时打包fat jar,可以使用provided风格
2、完全移除了对akka的依赖
3、mesos粗粒度模式下,支持启动多个executor
4、支持kryo 3.0版本
5、使用scala 2.11替代了scala 2.10

移除的功能

1、对hadoop 2.1以及之前版本的支持
2、spark 1.x中标识为过期的所有api
3、 hash-based shuffle manager
4、dataframe不再是一个类,而是dataset[Row]的类型别名
5、 bagel模块
6、闭包序列化配置的支持
7、HTTPBroadcast支持
8、基于TTL模式的元数据清理支持
9、半私有的org.apache.spark.Logging的使用支持
10、SparkContext.metricsSystem API
11、与tachyon的面向block的整合支持
12、 python dataframe中返回rdd的方法
13、使用很少的streaming数据源支持:twitter、akka、MQTT、ZeroMQ
14、standalone master的历史数据支持功能

变化的机制

1、要求基于scala 2.11版本进行开发,而不是scala 2.10版本
2、kryo版本升级到了3.0
3、SQL中的浮点类型,使用decimal类型来表示,而不是double类型
4、java的flatMap和mapPartitions方法,从iterable类型转变为iterator类型
5、java的countByKey返回<K,Long>类型,而不是<K,Object>类型
6、写parquet文件时,summary文件默认不会写了,需要开启参数来启用
7、spark mllib中,基于dataframe的api完全依赖于自己,不再依赖mllib包

Spark 1.x:Spark Core(RDD)、Spark SQL(SQL+Dataframe+Dataset)、Spark Streaming、Spark MLlib、Spark Graphx

Spark 2.x:Spark Core(RDD)、Spark SQL(ANSI-SQL+Subquery+Dataframe/Dataset)、Spark Streaming、Structured Streaming、Spark MLlib(Dataframe/Dataset)、Spark Graphx、Second Generation Tungsten Engine(Whole-stage code generation+Vectorization)

8、Spark 1.x到Spark 2.x,完全是一脉相承的关系,即,Spark 2.x基本上是基于Spark 1.x进行了更多的功能和模块的扩展,以及底层性能的改良。绝对不是说,Spark 2.x彻底淘汰和替代了Spark 1.x中的组件。而且实际上,对于Spark 1.x中90%以上的东西,Spark 2.x几乎都完全保留了支持和延续,并没有做任何改变。

9、从Spark诞生之日开始,RDD就是Spark最主要的编程接口,重要程度类似于Hadoop中的MapReduce。RDD,简单来说,就是一个不可变的分布式数据集,被分为多个partition从而在一个集群上分布式地存储。我们可以使用RDD提供的各种transformation和action算子,对RDD执行分布式的计算操作。可能很多人会问,Spark 2.0开始,包括Structured Streaming、Spark MLlib、Spark SQL底层都开始基Dataframe/Dataset来作为基础计算引擎,那么Spark Core/RDD是不是就要被淘汰了?
Spark Core绝对不会被淘汰掉。因为Spark Core/RDD作为一种low-level的API有它的较为底层的应用场景,虽然后续这种场景会越来越少,Dataframe/Dataset API会逐渐替代原先Spark Core的一些场景,但是不可否认的是,这种场景还是存在的。此外,Dataframe/Dataset实际上底层也是基于Spark Core/RDD构建的。所以说,Spark Core/RDD是Spark生态中,不可替代的基础API和引擎,其他所有的组件几乎都是构建在它之上。未来它不会被淘汰,只是应用场景会减少而已。

10、Spark 2.x中,在离线批处理计算中,编程API,除了RDD以外,还增强了Dataframe/Dataset API。那么,我们到底什么时候应该使用Spark Core/RDD来进行编程呢?实际上,RDD和Dataset最大的不同在于,RDD是底层的API和内核,Dataset实际上基于底层的引擎构建的high-level的计算引擎。

RDD应用场景

1、如果我们需要对数据集进行非常底层的掌控和操作,比如说,手动管理RDD的分区,或者根据RDD的运行逻辑来结合各种参数和编程来进行较为底层的调优。因为实际上Dataframe/Dataset底层会基于whole-stage code generation技术自动生成很多代码,那么就意味着,当我们在进行线上报错的troubleshooting以及性能调优时,对Spark的掌控能力就会降低。而使用Spark Core/RDD,因为其运行完全遵循其源码,因此我们完全可以在透彻阅读Spark Core源码的基础之上,对其进行troubleshooting和底层调优。(最重要的一点)
2、我们要处理的数据是非结构化的,比如说多媒体数据,或者是普通文本数据。
3、我们想要使用过程式编程风格来处理数据,而不想使用domain-specific language的编程风格来处理数据。
4、我们不关心数据的schema,即元数据。
5、我们不需要Dataframe/Dataset底层基于的第二代tungsten引擎提供的whole-stage code generation等性能优化技术。

11、就像RDD一样,Dataframe也代表一个不可变的分布式数据集。与RDD不同的一点是,Dataframe引入了schema的概念,支持以复杂的类型作为元素类型,同时指定schema,比如Row。因此Dataframe更像是传统关系型数据库中的表的概念。为了提升开发人员对大数据的处理能力,Dataframe除了提供schema的引入,还基于Schema提供了很多RDD所不具备的high-level API,以及一些domain-specific language(特定领域编程语言)。但是在Spark 2.0中,Dataframe和Dataset合并了,Dataframe已经不是一个单独的概念了,目前仅仅只是Dataset[Row]的一个类型别名而已,你可以理解为Dataframe就是Dataset。
在这里插入图片描述
从Spark 2.0开始,Dataset有两种表现形式:typed API和untyped API。我们可以认为,Dataframe就是Dataset[Row]的别名,Row就是一个untyped类型的对象,因为Row是类似于数据库中的一行,我们只知道里面有哪些列,但是有些列即使不存在,我们也可以这对这些不存在的列进行操作。因此其被定义为untyped,就是弱类型。

而Dataset[T]本身,是一种typed类型的API,其中的Object通常都是我们自己自定义的typed类型的对象,因为对象是我们自己定义的,所以包括字段命名以及字段类型都是强类型的。目前Scala支持Dataset和Dataframe两种类型,Java仅仅支持Dataset类型,Python和R因为不具备compile-time type-safety特性,因此仅仅支持Dataframe。

Dataset API优点

1、静态类型以及运行时的类型安全性

SQL语言具有最不严格的限制,而Dataset具有最严格的限制。SQL语言在只有在运行时才能发现一些错误,比如类型错误,但是由于Dataframe/Dataset目前都是要求类型指定的(静态类型),因此在编译时就可以发现类型错误,并提供运行时的类型安全。比如说,如果我们调用了一个不属于Dataframe的API,编译时就会报错。但是如果你使用了一个不存在的列,那么也只能到运行时才能发现了。而最严格的就是Dataset了,因为Dataset是完全基于typed API来设计的,类型都是严格而且强类型的,因此如果你使用了错误的类型,或者对不存在的列进行了操作,都能在编译时就发现。

2、将半结构化的数据转换为typed自定义类型

举例来说,如果我们现在有一份包含了学校中所有学生的信息,是以JSON字符串格式定义的,比如:{“name”: “leo”, “age”, 19, “classNo”: 1}。我们可以自己定义一个类型,比如case class Student(name: String, age: Integer, classNo: Integer)。接着我们就可以加载指定的json文件,并将其转换为typed类型的Dataset[Student],比如val ds = spark.read.json(“students.json”).as[Student]。

在这里,Spark会执行三个操作:
1、Spark首先会读取json文件,并且自动推断其schema,然后根据schema创建一个Dataframe。
2、在这里,会创建一个Dataframe=Dataset[Row],使用Row来存放你的数据,因为此时还不知道具体确切的类型。
3、接着将Dataframe转换为Dataset[Student],因为此时已经知道具体的类型是Student了。

这样,我们就可以将半结构化的数据,转换为自定义的typed结构化强类型数据集。并基于此,得到之前说的编译时和运行时的类型安全保障。

3、API的易用性

Dataframe/Dataset引入了很多的high-level API,并提供了domain-specific language风格的编程接口。这样的话,大部分的计算操作,都可以通过Dataset的high-level API来完成。通过typed类型的Dataset,我们可以轻松地执行agg、select、sum、avg、map、filter、groupBy等操作。使用domain-specific language也能够轻松地实现很多计算操作,比如类似RDD算子风格的map()、filter()等。

4、性能

除了上述的优点,Dataframe/Dataset在性能上也有很大的提升。首先,Dataframe/Dataset是构建在Spark SQL引擎之上的,它会根据你执行的操作,使用Spark SQL引擎的Catalyst来生成优化后的逻辑执行计划和物理执行计划,可以大幅度节省内存或磁盘的空间占用的开销(相对于RDD来说,Dataframe/Dataset的空间开销仅为1/3~1/4),也能提升计算的性能。其次,Spark 2.x还引入第二代Tungsten引擎,底层还会使用whole-stage code generation、vectorization等技术来优化性能。

什么时候应该使用Dataframe/Dataset,而不是RDD呢?

1、如果需要更加丰富的计算语义,high-level的抽象语义,以及domain-specific API。
2、如果计算逻辑需要high-level的expression、filter、map、aggregation、average、sum、SQL、列式存储、lambda表达式等语义,来处理半结构化,或结构化的数据。
3、如果需要高度的编译时以及运行时的类型安全保障。
4、如果想要通过Spark SQL的Catalyst和Spark 2.x的第二代Tungsten引擎来提升性能。
5、如果想要通过统一的API来进行离线、流式、机器学习等计算操作。
6、如果是R或Python的用户,那么只能使用Dataframe。


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇Spark编程指南(三) 下一篇spark2.0.2安装记录

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(214) }