数据框

使用像 DataFrame 这样的结构化 API 而不是 RDD 时，开发者不仅能获得更好的性能，还能体验到更高的表达性、简洁性、可组合性和一致性。结构化 API（包括 DataFrame 和 Dataset）使得编写数据分析中常见的计算模式更加直观。以下是一个使用 RDD 计算三名分析师对电影《乱世佳人》评分平均值的示例代码：

1sc.parallelize(Seq(("Gone with the Wind", 6), ("Gone with the Wind", 8),
2  ("Gone with the Wind", 8)))
3  .map(v => (v._1, (v._2, 1)))
4  .reduceByKey((k, v) => (k._1 + v._1, k._2 + v._2))
5  .map(x => (x._1, x._2._1 / (x._2._2 * 1.0)))
6  .collect()

这段代码显得晦涩难懂。从 Spark 的角度看，它无法理解 lambda 函数中的意图，也无法进行任何优化。相比之下，使用 DataFrame 实现相同计算的代码如下：

1val dataDF = spark.createDataFrame(Seq(("Gone with the Wind", 6),
2  ("Gone with the Wind", 8), ("Gone with the Wind", 8))).toDF("movie", "rating")
3val avgDF = dataDF.groupBy("movie").agg(avg("rating"))
4avgDF.show()

对比两段代码可以明显看出，使用 DataFrame 的版本更具可读性和表达力。与 RDD 版本不同，DataFrame 版本告诉 Spark 我们想要做什么，而不是 如何去做。例如，使用 DataFrame 时，我们可以直接指示 Spark 计算平均值；而使用 RDD 时，我们必须手动写出计算平均值的逻辑。此外，使用高级 API 的另一个优势是，在支持的语言之间代码风格保持一致和统一。如果我们将上述 DataFrame 代码用 Python 重写，其结构将与 Scala 版本非常相似，但若使用 RDD，则可能完全不同。

DataFrame

DataFrame 是最常用的结构化 API，代表一个具有行和列的表格。每一列都有由模式定义的数据类型。你可以将 DataFrame 理解为一张超大表格，无法存储在单台机器上，因此被分割并分布在整个集群的多台机器上。即使这张表格能放入单台机器，但所需计算耗时过长，也必须将数据分块并在多台机器上并行处理。

另一种理解 DataFrame 的方式是：它们是分布式、类似表格的数据集合，具有明确定义的行和列。每列的数据类型在所有行中保持一致。从本质上讲，DataFrame（以及稍后介绍的 Dataset）是用于在集群中多台机器间执行数据操作的惰性求值计划。

DataFrame 被划分为更小的部分，称为 分区。 一个分区是父 DataFrame 中位于集群某台物理机器上的若干行数据集合。DataFrame 的分区决定了数据在集群中的物理分布方式。分区数量也决定了 Spark 作业能够实现的并行度。如果只有一个分区，即使有数百个可用的执行器，也只能由一个执行器处理数据；反之，如果有大量分区但仅有一个执行器可用，则也无法实现并行处理。

在使用 DataFrame 时，用户不会手动或逐个操作分区。相反，用户指定高层次的数据转换操作，Spark 框架会自动将这些操作应用到集群中所有分区上。

模式

模式定义了 DataFrame 的列名和数据类型。模式可以手动定义，也可以从数据源中读取。Spark 支持 模式推断。Spark 会读取部分数据行，并解析其中的类型，将其映射为对应的 Spark 类型。我们还可以通过 schema 方法查看某个 DataFrame 对象的推断模式。

Spark 类型

Spark 使用名为 Tungsten 的引擎来维护内部类型信息。Spark 类型与支持语言（如 Java 或 Python）中的类型相对应。Spark 会将用任一支持语言编写的表达式转换为等效的 Tungsten 表示形式。Tungsten 引擎应用多种优化，并持续改进以提升执行效率。

在 Scala 中，我们可以将 DataFrame 的列定义为 String、Byte、Long、Map 等类型。Scala 中可用的基本类型如下：

所有基本类型都是 DataTypes 的子类型，除了 DecimalType。Spark 也为 Python 提供了类似的类型支持。显然，数据分析往往需要超出基本类型的复杂数据结构，Spark 在 Scala 和 Python 中均提供了强大的复合类型支持，如 Map、Array、Struct、Date、Timestamp 以及字段等。具体如下：

本课程中使用的所有查询和命令均已复制到下方的小工具中，方便复制粘贴到终端中使用。

1# 使用 RDD 计算平均评分
2(sc.parallelize(Seq(("Gone with the Wind", 6), ("Gone with the Wind", 8),
3  ("Gone with the Wind", 8)))
4  .map(v => (v._1, (v._2, 1)))
5  .reduceByKey((k, v) => (k._1 + v._1, k._2 + v._2))
6  .map(x => (x._1, x._2._1 / (x._2._2 * 1.0)))
7  .collect())
8
9# 使用 DataFrame 计算平均评分
10val dataDF = (spark.createDataFrame(Seq(("Gone with the Wind", 6),
11  ("Gone with the Wind", 8), ("Gone with the Wind", 8)))
12  .toDF("movie", "rating"))
13
14val avgDF = (dataDF.groupBy("movie").agg(avg("rating")))
15avgDF.show()