sparksql DataSet和DataFrame

Dataset是分布式数据集合。Dataset是Spark1.6新增的接口，用以提供RDDs（强类型，有使用强大的lambda函数的能力）的优点和Spark SQL的经优化的执行引擎的优点。Dataset可以从JVM对象进行构造并通过转换函数（如map，flatmap，filter等）进行操作。DatasetAPI支持Scala和Java。Python不支持Dataset API。但因为Python本身的动态性，DatasetAPI的许多优点都已经可用（比如，你可以通过名字很自然的访问一行的某一个字段，如row.columnName），R的情况与此类似。

Dataset与RDD很像，不同的是它并不使用Java序列化或者Kryo，而是使用特殊的编码器来为网络间的处理或传输的对象进行序列化。对转换一个对象为字节的过程来说编码器和标准系列化器都是可靠的，编码器的代码是自动生成并且使用了一种格式，这种格式允许Spark在不需要将字节解码成对象的情况下执行很多操作，如filtering、sorting和hashing等。

DataFrame是Dataset组织成列的数据集。它在概念上相当于关系型数据库中的表，或者R/Python中的数据帧，但是在底层进行了更多的优化。DataFrames可以从多种数据源创建，例如：结构化数据文件、Hive中的表、外部数据库或者已存在的RDDs。DataFrame API支持Scala、Java、Python和R。在Scala和Java中DataFrame其实是Dataset的RowS的形式的表示。在Scala API中，DataFrame仅仅是Dataset[Row]的别名。但在Java中，使用者需要使用Dataset来表示一个DataFrame。

但是 DataFrame 出来后发现有些情况下 RDD 可以表达的逻辑用 DataFrame 无法表达。比如 要对 group by 或 join 后的结果用自定义的函数,可能用 SQL 是无法表达的。如下代码：

case class ClassData(a: String, b: Int)

case class ClassNullableData(a: String, b: Integer)

val ds = Seq(ClassData("a", 1), ClassData("a", 2)).toDS()

val agged = ds.groupByKey(d => ClassNullableData(d.a, null))

.mapGroups {

case (key, values) => key.a + values.map(_.b).sum

}

中间处理过程的数据是自定义的类型,并且 groupby 后的聚合逻辑也是自定义的，故用 SQL 比较难以表达，所以提出了 Dataset API。Dataset API 扩展 DataFrame API 支持静态类型和运行已经存在的 Scala 或 Java 语言的用户自定义函数。同时 Dataset 也能享受 Spark SQL 里所有性能 带来的提升。

Spark SQL的Scala接口支持自动的将一个包含case class的RDD转换为DataFrame。这个case class定义了表结构。Caseclass的参数名是通过反射机制读取，然后变成列名。Caseclass可以嵌套或者包含像Seq或Array之类的复杂类型。这个RDD可以隐式的转换为一个DataFrame，然后被注册为一张表。这个表可以随后被SQL的statement使用。

Spark SQL支持两种将已存在的RDD转化为Dataset的方法。第一种方法使用反射推断包含特定类型对象的RDD的结构。这种基于反射的方法代码更加简洁，并且当你在写Spark程序的时候已经知道RDD的结构的情况下效果很好。

第二种创建Dataset的方法是通过编程接口建立一个结构，然后将它应用于一个存在的RDD。虽然这种方法更加繁琐，但它允许你在运行之前不知道其中的列和对应的类型的情况下构建Dataset。

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