Catalyst Optimizer:
Dataset
数据集仅可用Scala或Java。但是，我们提供了以下上下文来更好地理解Spark 2.0的方向数据集是在2015年作为Apache Spark 1.6版本的一部分引入的。datasets的目标是提供一个类型安全的编程接口。
这允许开发人员使用具有编译时类型安全性的半结构化数据(如JSON或键值对)进行工作(也就是说，生产应用程序在运行之前可以检查错误)。
Python不实现Dataset API的部分原因是Python不是一种类型安全的语言。同样重要的是，数据集API包含高级领域特定的语言操作，如sum()、avg()、join()和group()。
后一个特性意味着您具有传统Spark RDDs的灵活性，但是代码也更容易表达、读取和编写。
从下图中可以看出，DataFrame和Dataset都属于作为Apache Spark的一部分引入的新Dataset API2.0:
DataFrame和Dataset api的统一有可能创建打破向后兼容性的更改。
从下图中可以看出，DataFrame和Dataset都属于作为Apache Spark的一部分引入的新Dataset API2.0:DataFrame和Dataset api的统一有可能创建打破向后兼容性的更改。
这是Apache Spark 2.0成为主要版本的主要原因之一(最小化任何中断的更改)。从下图中可以看出，DataFrame和Dataset都属于作为Apache Spark的一部分引入的新Dataset API2.0:
如前所述，Dataset API提供了一个类型安全的、面向对象的编程接口。数据集可以通过将表达式和数据字段公开给查询接口和
的快速内存编码来利用
。但是，随着DataFrame和Dataset现在作为Apache Spark 2.0的一部分统一起来，DataFrame现在是Dataset非类型化API的别名。
更具体地说:
DataFrame = Dataset[T]
DataFrame是什么？
在Spark中，DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库中的二维表格。DataFrame与RDD的主要区别在于，前者带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得Spark SQL得以洞察更多的结构信息，从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。反观RDD，由于无从得知所存数据元素的具体内部结构，Spark Core只能在stage层面进行简单、通用的流水线优化。
创建DataFrame
在Spark SQL中，开发者可以非常便捷地将各种内、外部的单机、分布式数据转换为DataFrame。以下Python示例代码充分体现了Spark SQL 1.3.0中DataFrame数据源的丰富多样和简单易用：
# 从Hive中的users表构造DataFrame users = sqlContext.table("users") # 加载S3上的JSON文件 logs = sqlContext.load("s3n://path/to/data.json", "json") # 加载HDFS上的Parquet文件 clicks = sqlContext.load("hdfs://path/to/data.parquet", "parquet") # 通过JDBC访问MySQL comments = sqlContext.jdbc("jdbc:mysql://localhost/comments", "user") # 将普通RDD转变为DataFrame rdd = sparkContext.textFile("article.txt") \ .flatMap(lambda line: line.split()) \ .map(lambda word: (word, 1)) \ .reduceByKey(lambda a, b: a + b) \ wordCounts = sqlContext.createDataFrame(rdd, ["word", "count"]) # 将本地数据容器转变为DataFrame data = [("Alice", 21), ("Bob", 24)] people = sqlContext.createDataFrame(data, ["name", "age"]) # 将Pandas DataFrame转变为Spark DataFrame（Python API特有功能） sparkDF = sqlContext.createDataFrame(pandasDF)
使用DataFrame
和Pandas类似，Spark DataFrame也提供了一整套用于操纵数据的DSL。
这些DSL在语义上与SQL关系查询非常相近（这也是Spark SQL能够为DataFrame提供无缝支持的重要原因之一）。以下是一组用户数据分析示例：
# 创建一个只包含"年轻"用户的DataFrame young = users.filter(users.age < 21) # 也可以使用Pandas风格的语法 young = users[users.age < 21] # 将所有人的年龄加1 young.select(young.name, young.age + 1) # 统计年轻用户中各性别人数 young.groupBy("gender").count() # 将所有年轻用户与另一个名为logs的DataFrame联接起来 young.join(logs, logs.userId == users.userId, "left_outer")
除DSL以外，我们当然也可以像以往一样，用SQL来处理DataFrame：
young.registerTempTable("young") sqlContext.sql("SELECT count(*) FROM young")
最后，当数据分析逻辑编写完毕后，我们便可以将最终结果保存下来或展现出来：
# 追加至HDFS上的Parquet文件 young.save(path="hdfs://path/to/data.parquet", source="parquet", mode="append") # 覆写S3上的JSON文件 young.save(path="s3n://path/to/data.json", source="json", mode="append") # 保存为SQL表 young.saveAsTable(tableName="young", source="parquet" mode="overwrite") # 转换为Pandas DataFrame（Python API特有功能） pandasDF = young.toPandas() # 以表格形式打印输出 young.show()</span>
1.幕后英雄：Spark SQL查询优化器与代码生成
正如RDD的各种变换实际上只是在构造RDD DAG，DataFrame的各种变换同样也是lazy的。它们并不直接求出计算结果，而是将各种变换组装成与RDD DAG类似的逻辑查询计划。如前所述，由于DataFrame带有schema元信息，Spark SQL的查询优化器得以洞察数据和计算的精细结构，从而施行具有很强针对性的优化。随后，经过优化的逻辑执行计划被翻译为物理执行计划，并最终落实为RDD DAG。
这样做的好处体现在几个方面：
1. 用户可以用更少的申明式代码阐明计算逻辑，物理执行路径则交由Spark SQL自行挑选。一方面降低了开发成本，一方面也降低了使用门槛——很多情况下，即便新手写出了较为低效的查询，Spark SQL也可以通过过滤条件下推、列剪枝等策略予以有效优化。这是RDD API所不具备的。
2. Spark SQL可以动态地为物理执行计划中的表达式生成JVM字节码，进一步实现归避虚函数调用开销、削减对象分配次数等底层优化，使得最终的查询执行性能可以与手写代码的性能相媲美。
3. 对于PySpark而言，采用DataFrame编程时只需要构造体积小巧的逻辑执行计划，物理执行全部由JVM端负责，Python解释器和JVM间大量不必要的跨进程通讯得以免除。如上图所示，一组简单的对一千万整数对做聚合的测试中，PySpark中DataFrame API的性能轻松胜出RDD API近五倍。此外，今后Spark SQL在Scala端对查询优化器的所有性能改进，PySpark都可以免费获益。
2.外部数据源API增强
数据写入支持
在Spark 1.2.0中，外部数据源API只能将外部数据源中的数据读入Spark，而无法将计算结果写回数据源；同时，通过数据源引入并注册的表只能是临时表，相关元信息无法持久化。在1.3.0中，我们提供了完整的数据写入支持，从而补全了多数据源互操作的最后一块重要拼图。前文示例中Hive、Parquet、JSON、Pandas等多种数据源间的任意转换，正是这一增强的直接成果。
站在Spark SQL外部数据源开发者的角度，数据写入支持的API主要包括：
2.1. 数据源表元数据持久化
CREATE [TEMPORARY] TABLE [IF NOT EXISTS] <table-name> [(col-name data-type [, ...)] USING <source> [OPTIONS ...] [AS <select-query>]
由此，注册自外部数据的SQL表既可以是临时表，也可以被持久化至Hive metastore。需要持久化支持的外部数据源，除了需要继承原有的RelationProvider以外，还需继承CreatableRelationProvider。
2.2. InsertableRelation
支持数据写入的外部数据源的relation类，还需继承trait InsertableRelation，并在insert方法中实现数据插入逻辑。
Spark 1.3.0中内置的JSON和Parquet数据源都已实现上述API，可以作为开发外部数据源的参考示例。
3.统一的load/save API
在Spark 1.2.0中，要想将SchemaRDD中的结果保存下来，便捷的选择并不多。常用的一些包括：
rdd.saveAsParquetFile(...)
rdd.saveAsTextFile(...)
rdd.toJSON.saveAsTextFile(...)
rdd.saveAsTable(...)
....
可见，不同的数据输出方式，采用的API也不尽相同。更令人头疼的是，我们缺乏一个灵活扩展新的数据写入格式的方式。
针对这一问题，1.3.0统一了load/save API，让用户按需自由选择外部数据源。这套API包括：
1.SQLContext.table #从SQL表中加载DataFrame。 2.SQLContext.load #从指定的外部数据源加载DataFrame。 3.SQLContext.createExternalTable #将指定位置的数据保存为外部SQL表，元信息存入Hive metastore，并返回包含相应数据的DataFrame。 4.DataFrame.save #将DataFrame写入指定的外部数据源。 5.DataFrame.saveAsTable #将DataFrame保存为SQL表，元信息存入Hive metastore，同时将数据写入指定位置。
4.Parquet数据源增强
Spark SQL从一开始便内置支持Parquet这一高效的列式存储格式。在开放外部数据源API之后，原有的Parquet支持也正在逐渐转向外部数据源。1.3.0中，Parquet外部数据源的能力得到了显著增强。主要包括schema合并和自动分区处理。
1.Schema合并
与ProtocolBuffer和Thrift类似，Parquet也允许用户在定义好schema之后随时间推移逐渐添加新的列，只要不修改原有列的元信息，新旧schema仍然可以兼容。这一特性使得用户可以随时按需添加新的数据列，而无需操心数据迁移。
2.分区信息发现
按目录对同一张表中的数据分区存储，是Hive等系统采用的一种常见的数据存储方式。新的Parquet数据源可以自动根据目录结构发现和推演分区信息。
3.分区剪枝
分区实际上提供了一种粗粒度的索引。当查询条件中仅涉及部分分区时，通过分区剪枝跳过不必要扫描的分区目录，可以大幅提升查询性能。
以下Scala代码示例统一展示了1.3.0中Parquet数据源的这几个能力（Scala代码片段）：
// 创建两个简单的DataFrame，将之存入两个独立的分区目录 val df1 = (1 to 5).map(i => (i, i * 2)).toDF("single", "double") df1.save("data/test_table/key=1", "parquet", SaveMode.Append)
val df2 = (6 to 10).map(i => (i, i * 2)).toDF("single", "double") df2.save("data/test_table/key=2", "parquet", SaveMode.Append)
// 在另一个DataFrame中引入一个新的列，并存入另一个分区目录 val df3 = (11 to 15).map(i => (i, i * 3)).toDF("single", "triple") df3.save("data/test_table/key=3", "parquet", SaveMode.Append)
// 一次性读入整个分区表的数据 val df4 = sqlContext.load("data/test_table", "parquet") // 按分区进行查询，并展示结果 val df5 = df4.filter($"key" >= 2)
df5.show()
这段代码的执行结果为：
6 12 null 2 7 14 null 2 8 16 null 2 9 18 null 2 10 20 null 2 11 null 33 3 12 null 36 3 13 null 39 3 14 null 42 3 15 null 45 3
可见，Parquet数据源自动从文件路径中发现了key这个分区列，并且正确合并了两个不相同但相容的schema。值得注意的是，在最后的查询中查询条件跳过了key=1这个分区。Spark SQL的查询优化器会根据这个查询条件将该分区目录剪掉，完全不扫描该目录中的数据，从而提升查询性能。
总体来说
- schema : RDD每一行的数据, 结构都是一样的. 这个结构就存储在schema中. Spark通过schame就能够读懂数据, 因此在通信和IO时就只需要序列化和反序列化数据, 而结构的部分就可以省略了.
-
off-heap : 意味着JVM堆以外的内存, 这些内存直接受操作系统管理（而不是JVM）。Spark能够以二进制的形式序列化数据(不包括结构)到off-heap中, 当要操作数据时, 就直接操作off-heap内存. 由于Spark理解schema, 所以知道该如何操作.
off-heap就像地盘, schema就像地图, Spark有地图又有自己地盘了, 就可以自己说了算了, 不再受JVM的限制, 也就不再收GC的困扰了.
通过schema和off-heap, DataFrame解决了RDD的缺点, 但是却丢了RDD的优点. DataFrame不是类型安全的, API也不是面向对象风格的.所以我们后来在spark2.0引入的dataset。