Spark权威指南(中文版)----第5章 结构化API基本操作

Spark The Definitive Guide(Spark权威指南) 中文版。本书详细介绍了Spark2.x版本的各个模块，目前市面上最好的Spark2.x学习书籍！！！

扫码关注公众号:登峰大数据，阅读中文Spark权威指南(完整版)，系统学习Spark大数据框架！

在第4章中，我们介绍了结构化API的核心抽象。本章将从架构概念转向您将使用的工具来操作DataFrames和其中的数据。本章专门讨论基本的DataFrame操作。聚合操作、窗口函数操作和连接操作将在以后的章节中讨论。

API查看地址：https://spark.apache.org/docs/2.2.0/api/scala/#org.apache.spark.sql.Dataset

从定义上看，一个DataFrame包括一系列的records（记录。就像table中的rows）,这些行的类型是Row类型，包括一系列的columns（就像电子表格中的列。），作用于数据集每条记录之上的计算表达式，实际上是作用于数据记录中的columns之上。Schema定义了每一列的列名和数据类型。DataFrame的分区定义了DataFrame或Dataset在整个集群中的物理分布情况。

让我们通过一个例子来看下如何创建一个DataFrame:

// in Scala
val df = spark.read.format("json").load("/data/flight-data/json/2015-summary.json")

# in Python
df = spark.read.format("json").load("/data/flight-data/json/2015-summary.json")

我们讨论了DataFame包含列，并且我们使用schema来定义它们。让我们看一下当前DataFrame上的schema:

df.printSchema()

schema将一切联系在一起，所以它们值得我们深入讨论。

5.1.  Schemas

schema定义了DataFrame的列名和数据类型。我们可以让数据源定义模式(称为schema-on-read)，或者我们可以自己定义它。

警告
在读取数据之前，是否需要定义schema取决于您的用例场景。对于某些特殊的分析，schema-on-read通常是很适合的(尽管有时读取文本格式数据，它可能会慢一些，比如CSV或JSON)。但是，这也会导致精度问题，比如在读取文件时，Long类型错误地设置为Integer类型。当使用Spark进行生产提取、转换和加载(ETL)时，手动定义模式通常是个好主意，特别是在使用诸如CSV和JSON之类的非类型化数据源时，因为schema infer（模式推理）可能根据所读数据的类型而变化。

让我们从一个简单的文件开始，这是我们在第4章中看到的，让半结构化的Json文件特性来定义这个schema。这是美国运输统计局的航班数据:

// in Scala
spark.read.format("json").load("/data/flight-data/json/2015-summary.json").schema

返回结果如下：

org.apache.spark.sql.types.StructType = ...
StructType(StructField(DEST_COUNTRY_NAME,StringType,true),
StructField(ORIGIN_COUNTRY_NAME,StringType,true),
StructField(count,LongType,true))

一个schema就是一个StructType，由多个StructField类型的fields组成，每个field包括一个列名称、一个列类型、一个布尔型的标识（是否可以有缺失值和null值）。

schema可以包含其他StructType(Spark的复杂类型)。在第6章中，我们将讨论复杂类型。如果数据中的类型(在运行时)与模式不匹配，Spark将抛出一个错误。下面的示例展示了如何在DataFrame上创建和执行特定的schema。

// in Scala
import org.apache.spark.sql.types.{StructField, StructType, StringType, LongType}
import org.apache.spark.sql.types.Metadata

val myManualSchema = StructType(Array(
 StructField("DEST_COUNTRY_NAME", StringType, true),
 StructField("ORIGIN_COUNTRY_NAME", StringType, true),
 StructField("count", LongType, false,
   Metadata.fromJson("{"hello":"world"}"))
))

val df = spark.read.format("json").schema(myManualSchema)
.load("/data/flight-data/json/2015-summary.json")

正如第4章所讨论的，我们不能简单地通过每种语言类型设置类型，因为Spark维护了它自己的类型信息。现在让我们讨论一下scheme定义的内容：columns。

5.2.  列和表达式(Columns 和 Expressions)

Spark中的列类似于电子表格中的列。您可以从DataFrame中选择列、操作列和删除列，这些操作称为Expressions表达式。

对Spark来说，列是逻辑结构，它仅仅表示通过一个表达式按每条记录计算出的一个值。这意味着，要得到一个column列的真实值，我们需要有一行row数据，为了得到一行数据，我们需要有一个DataFrame。您不能在DataFrame的上下文之外操作单个列。您必须在DataFrame内使用Spark转换来修改列的内容。

5.2.1.   Columns

有许多不同的方法来构造和引用列，但最简单的两种方法是使用col ( ) 或 column ( ) 函数。要使用这些函数中的任何一个，您需要传入一个列名:

// in Scala
import org.apache.spark.sql.functions.{col, column}
col("someColumnName")
column("someColumnName")

在这本书中我们将使用col( )函数。如前所述，这个列可能存在于我们的DataFrames中，也可能不存在。在将列名称与我们在Catalog中维护的列进行比较之前，列不会被解析，即列是unresolved。列和表解析发生在analyzer分析器阶段，如第4章所述。

注意
我们刚才提到的两种不同的方法引用列。Scala有一些独特的语言特性，允许使用更多的简写方式来引用列。以下的语法糖执行完全相同的事情，即创建一个列，但不提供性能改进:
$"myColumn"
'myColumn

$允许我们将一个字符串指定为一个特殊的字符串，该字符串应该引用一个表达式。标记(')是一种特殊的东西，称为符号; 这是一个特定于scala语言的，指向某个标识符。它们都执行相同的操作，是按名称引用列的简写方式。当您阅读不同的人的Spark代码时，您可能会看到前面提到的所有引用。我们把选择权留给您，您可以使用任何对您和您工作的人来说最舒适和最容易维护的东西。

显式列引用

如果需要引用特定的DataFrame的列，可以在特定的DataFrame上使用col方法。当您执行连接时，这可能非常有用，并且需要引用一个DataFrame中的特定列，该列可能与连接的DataFrame中的另一个列共享一个名称。

5.2.2.   表达式Expressions

我们前面提到过，列是表达式，但表达式是什么? 表达式是在DataFrame中数据记录的一个或多个值上的一组转换。把它想象成一个函数，它将一个或多个列名作为输入，解析它们，然后潜在地应用更多的表达式，为数据集中的每个记录创建一个单一值。重要的是，这个“单一值”实际上可以是一个复杂的类型，比如映射或数组。我们将在第六章中看到更多的复杂类型。

在最简单的情况下，通过expr函数创建的表达式只是一个DataFrame列引用。在最简单的情况下，expr(“someCol”)等价于col(“someCol”)。

列可以作为表达式

列提供了表达式功能的一个子集。如果您使用col()并希望在该列上执行转换，则必须在该列引用上执行转换。在使用表达式时，expr函数实际上可以从字符串解析转换和列引用，并可以将其传递到进一步的转换中。让我们看一些例子。

expr(“someCol - 5”)与执行col(“someCol”)- 5，或甚至expr(“someCol”)- 5的转换相同。这是因为Spark将它们编译为一个逻辑树，逻辑树指定了操作的顺序。这一开始可能有点让人迷惑，但记住几个关键点:

• 列就是表达式

• 这些列上的列表达式和转换编译成与解析表达式相同的逻辑计划。

让我们以一个例子来说明:

(((col("someCol") + 5) * 200) - 6) < col("otherCol")

图5-1显示了该逻辑树的概述。

这看起来很熟悉，因为它是一个有向无环图。此图等价于以下代码:

// in Scala
import org.apache.spark.sql.functions.expr
expr("(((someCol + 5) * 200) - 6) < otherCol")

这是一个非常重要的观点。请注意前面的表达式实际上是有效的SQL代码，就像您可能放入SELECT语句一样? 这是因为这个SQL表达式和之前的DataFrame代码在执行之前编译到相同的底层逻辑树。这意味着您可以将表达式编写为DataFrame代码或SQL表达式，并获得完全相同的性能特征。

访问DataFrame中的columns

有时，您需要查看DataFrame的列，您可以使用类似printSchema的方法来实现; 但是，如果您想通过编程访问列，可以使用columns属性查看DataFrame上的所有列:

spark.read.format("json").load("/data/flight-data/json/2015-summary.json").columns

5.3.  Records 和 Rows

在Spark中，DataFrame中的每一行都是单个记录。Spark表示此记录为Row类型的对象。即一个record是一个Row类型的对象。Spark使用列表达式expression操作Row对象，以产生有效的结果值。Row对象的内部表示为：字节数组。因为我们使用列表达式操作Row对象，所以，字节数据不会对最终用户展示，即对用户不可见。

5.3.1.   创建Rows

您可以手动实例化一个Row对象，并实例化每一列的值。需要注意的是：只有DataFrame有schema，Row本身没有schema。这意味着，如果您手动创建一个Row对象，则必须以与它们可能被附加的DataFrame的schema相同的顺序指定列值(我们将在讨论创建DataFrames时看到这个):

// in Scala
import org.apache.spark.sql.Row
val myRow = Row("Hello", null, 1, false)

访问创建的Row对象中的数据非常容易: 你只需指定你想要列值的位置。

// in Scala
myRow(0) // type Any
myRow(0).asInstanceOf[String] // String
myRow.getString(0) // String
myRow.getInt(2) // Int

您还可以使用Dataset api在相应的Java虚拟机(JVM)对象中显式地返回一组数据（见第十一章）。

5.4.  DataFrame Transformations

DataFrame
https://spark.apache.org/docs/2.2.0/api/scala/#org.apache.spark.sql.functions$

现在我们简要地定义了DataFrame的核心部分，我们将开始操作DataFrames。在使用单一的DataFrames时，有一些基本的操作。这些分解为几个核心操作，如图5-2所示:

• 我们可以添加行或列

• 我们可以删除行或列

• 我们可以变换一行成一列(反之亦然)

• 我们可以根据列值对rows进行排序

幸运的是，我们可以将所有这些转换成简单的transformation方法，最常见的是那些使用一个列，逐行更改它，然后返回结果。

5.4.1.   创建DataFrame

如前所述，我们可以从原始数据源创建DataFrames。这在第9章中得到了广泛的讨论;但是，现在我们将使用它们来创建一个DataFrame示例。(在本章后面的演示目的中，我们还将把它注册为一个临时视图，以便我们可以用SQL查询它，并显示SQL中的基本转换。)

// in Scala
val df = spark.read.format("json")
.load("/data/flight-data/json/2015-summary.json")
df.createOrReplaceTempView("dfTable")
# in Python
df = spark.read.format("json").load("/data/flight-data/json/2015-summary.json")
df.createOrReplaceTempView("dfTable")

我们还可以创建动态DataFrames，通过转换一组数据为DataFrame。

// in Scala
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructField, StructType, StringType, LongType}

val myManualSchema = new StructType(Array(
 new StructField("some", StringType, true),
 new StructField("col", StringType, true),
 new StructField("names", LongType, false)))
val myRows = Seq(Row("Hello", null, 1L))
val myRDD = spark.sparkContext.parallelize(myRows)
val myDf = spark.createDataFrame(myRDD, myManualSchema)
myDf.show()

注意
在Scala中，我们还可以利用控制台中的Spark的implicits(如果您将它们导入到JAR代码中)，则可以在Seq类型上运行toDF。这不能很好地使用null类型，因此并不推荐用于生产用例。
// in Scala
val myDF = Seq(("Hello", 2, 1L)).toDF("col1", "col2", "col3")

现在，您已经知道如何创建DataFrames了，让我们来看看它们最有用的方法:

select方法：接收列column或表达式expression为参数。

selectExpr方法：接收字符串表达式expression为参数。

还有一些方法，通过函数的形式提供，在org.apache.spark.sql.functions包中。

使用这三个工具，您应该能够解决您在DataFrames中可能遇到的大多数数据分析挑战。

5.4.2.   select和selectExpr

select和selectExpr允许您在数据表上执行与SQL查询等效的DataFrame操作:

-- SQL查询方式
SELECT * FROM dataFrameTable
SELECT columnName FROM dataFrameTable
SELECT columnName * 10, otherColumn, someOtherCol as c FROM dataFrameTable

让我们浏览一些DataFrames的示例，讨论解决这个问题的一些不同方法。最简单的方法就是使用DataFrame的select方法，并将列名作为字符串参数：

// Scala方式
df.select("DEST_COUNTRY_NAME").show(2)
-- SQL方式
SELECT   DEST_COUNTRY_NAME    FROM    dfTable    LIMIT 2

您可以使用相同的查询样式选择多个列，只需在select方法调用中添加更多的列名字符串参数：

// in Scala
df.select("DEST_COUNTRY_NAME", "ORIGIN_COUNTRY_NAME").show(2)
-- in SQL
SELECT DEST_COUNTRY_NAME, ORIGIN_COUNTRY_NAME FROM dfTable LIMIT 2

正如在“列和表达式”章节中所讨论的，您可以用许多不同的方式引用列; 你需要记住的是，你可以交替使用它们:

// in Scala
import org.apache.spark.sql.functions.{expr, col, column}
df.select(
   df.col("DEST_COUNTRY_NAME"),
   col("DEST_COUNTRY_NAME"),
   column("DEST_COUNTRY_NAME"),
   'DEST_COUNTRY_NAME,
   $"DEST_COUNTRY_NAME",
   expr("DEST_COUNTRY_NAME"))
.show(2)

一个常见的错误是混合使用列对象和列字符串。例如，下列代码将导致编译错误:

df.select(col("DEST_COUNTRY_NAME"), "EST_COUNTRY_NAME")

正如我们到目前为止所看到的，expr是我们可以使用的最灵活的引用。它可以引用一个简单的列或一个列字符串操作。为了说明这一点，让我们更改列名，然后通过使用AS关键字来更改它。然后使用列上的alias方法:

// in Scala
df.select(expr("DEST_COUNTRY_NAME AS destination")).show(2)
# in Python
df.select(expr("DEST_COUNTRY_NAME AS destination")).show(2)
-- in SQL
SELECT DEST_COUNTRY_NAME as destination FROM dfTable LIMIT 2

这将列名更改为“destination”。您可以进一步操作您的表达式作为另一个表达式的结果:

// in Scala
df.select(expr("DEST_COUNTRY_NAME as destination").alias("DEST_COUNTRY_NAME")).show(2)

前面的操作将列名更改为原来的名称。

因为select后跟一系列expr是一种常见的模式，Spark有一个高效实现这一点的简写:selectExpr。这可能是日常使用中最方便的方式:

// in Scala
df.selectExpr("DEST_COUNTRY_NAME as newColumnName", "DEST_COUNTRY_NAME").show(2)

这开启了Spark的真正力量。我们可以将selectExpr视为一种构建复杂表达式的简单方法，这些表达式可以创建新的DataFrames。事实上，我们可以添加任何有效的非聚合SQL语句，只要能否进行列解析，它就有效。这里有一个简单的例子，它在我们的DataFrame中添加了一个新的列withinCountry，该列指定目的地和起点是否相同:

// in Scala
df.selectExpr(
   "*", // 包含所有的原始列
   "(DEST_COUNTRY_NAME = ORIGIN_COUNTRY_NAME) as withinCountry") .show(2)
-- in SQL
SELECT *, (DEST_COUNTRY_NAME = ORIGIN_COUNTRY_NAME) as withinCountry FROM dfTable LIMIT 2

输出为：

使用select expression，我们还可以利用我们拥有的函数来指定整个DataFrame上的聚合：

// in Scala
df.selectExpr("avg(count)", "count(distinct(DEST_COUNTRY_NAME))").show(2)
# in Python
df.selectExpr("avg(count)", "count(distinct(DEST_COUNTRY_NAME))").show(2)
-- in SQL
SELECT avg(count), count(distinct(DEST_COUNTRY_NAME)) FROM dfTable LIMIT 2

输出为：

5.4.3.   字面常量转换为Spark类型(Literals)

有时，我们需要将显式字面常量值传递给Spark，它只是一个值(而不是一个新列)。这可能是一个常数值或者我们以后需要比较的值。我们的方法是通过Literals，将给定编程语言的字面值转换为Spark能够理解的值。Literals是表达式，你可以用同样的方式使用它们:

// in Scala
import org.apache.spark.sql.functions.lit
df.select(expr("*"), lit(1).as("One")).show(2)
# in Python
from pyspark.sql.functions import lit
df.select(expr("*"), lit(1).alias("One")).show(2)

在SQL中，字面值只是特定的值:

-- in SQL
SELECT *, 1 as One FROM dfTable LIMIT 2

输出为：

当您可能需要检查一个值是否大于某个常量或其他通过编程创建的变量时，就会出现这种情况。

5.4.4.   添加列

还有一种更正式的方式，将新列添加到DataFrame中，这是通过在DataFrame上使用withColumn方法来实现的。例如，让我们添加一个列，将数字1添加为一个列:

// in Scala
df.withColumn("numberOne", lit(1)).show(2)
# in Python
df.withColumn("numberOne", lit(1)).show(2)
-- in SQL
SELECT *, 1 as numberOne FROM dfTable LIMIT 2

让我们做一些更有趣的事情，把它变成一个实际的表达式。在下一个示例中，我们将设置一个布尔标志，当出发国与目标国相同时:

// in Scala
df.withColumn("withinCountry", expr("ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME"))
.show(2)
# in Python
df.withColumn("withinCountry", expr("ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME"))
.show(2)

注意，withColumn函数有两个参数：列名和为DataFrame中的给定行创建值的表达式。有趣的是，我们也可以这样重命名列。SQL语法与前面的相同，因此我们可以在这个示例中省略它:

df.withColumn("Destination", expr("DEST_COUNTRY_NAME")).columns

结果为：

... DEST_COUNTRY_NAME, ORIGIN_COUNTRY_NAME, count, Destination

5.4.5.   重命名列

虽然我们可以按照刚才描述的方式重命名列，但是另一种方法是使用withcolumnrename方法。这会将第一个参数中的字符串的名称重命名为第二个参数中的字符串:

// in Scala
df.withColumnRenamed("DEST_COUNTRY_NAME", "dest").columns
... dest, ORIGIN_COUNTRY_NAME, count

5.4.6.   保留字和关键词

您可能遇到的一件事是保留字符，比如空格或列名称中的破折号。处理这些意味着适当地转义列名。在Spark中，我们通过使用单撇号(')字符来实现这一点。让我们使用withColumn，创建具有保留字符的列。我们将展示两个示例——在这里显示的一个示例中，我们不需要转义字符，但是在下一个示例中，我们需要:

// in Scala
import org.apache.spark.sql.functions.expr

val dfWithLongColName = df.withColumn("This Long Column-Name",
 expr("ORIGIN_COUNTRY_NAME"))

这里不需要转义字符，因为withColumn的第一个参数只是新列名的字符串。然而，在本例中，我们需要使用单撇号，因为我们在表达式中引用了一个列:

// in Scala
dfWithLongColName.selectExpr(
   "`This Long Column-Name`",
   "`This Long Column-Name` as `new col`")
.show(2)

dfWithLongColName.createOrReplaceTempView("dfTableLong")

-- in SQL
SELECT `This Long Column-Name`, `This Long Column-Name` as `new col`
FROM dfTableLong LIMIT 2

我们可以引用具有保留字符的列(而不是转义它们)，如果我们正在做显式的字符串到列的引用，它被解释为文字而不是表达式。我们只需要转义使用保留字符或关键字的表达式。下面两个例子从同一个DataFrame中都得到了结果:

// in Scala
dfWithLongColName.select(col("This Long Column-Name")).columns
# in Python
dfWithLongColName.select(expr("`This Long Column-Name`")).columns

5.4.7.   区分大小写

默认情况下，Spark是不区分大小写的;但是，您可以通过设置配置使Spark case变得大小写敏感:

-- in SQL
set spark.sql.caseSensitive true

5.4.8.   删除列

既然我们已经创建了这个列，那么让我们看看如何从DataFrames中删除列。您可能已经注意到我们可以通过使用select来实现这一点。然而，还有一个专门的方法叫做drop:

df.drop("ORIGIN_COUNTRY_NAME").columns

我们可以通过将多个列作为参数传递，来删除多个列:

dfWithLongColName.drop("ORIGIN_COUNTRY_NAME", "DEST_COUNTRY_NAME")

5.4.9.   更改列的类型(cast)

有时，我们可能需要列从一种类型转换到另一种类型;例如，如果我们有一组StringType应该是整数。我们可以将列从一种类型转换为另一种类型，例如，让我们将计数列从整数转换为类型Long:

df.withColumn("count2", col("count").cast("long"))
-- in SQL
SELECT *, cast(count as long) AS count2 FROM dfTable

5.4.10.   过滤行

为了过滤行，我们创建一个计算值为true或false的表达式。然后用一个等于false的表达式过滤掉这些行。使用DataFrames执行此操作的最常见方法是将表达式创建为字符串，或者使用一组列操作构建表达式。执行此操作有两种方法:您可以使用where或filter，它们都将执行相同的操作，并在使用DataFrames时接受相同的参数类型。因为熟悉SQL，我们将坚持使用where;但是，filter方法是有效的。

注意
在使用Scala或Java的Dataset API时，filter还接受Spark将应用于数据集中每个记录的任意函数。有关更多信息，请参见第11章。

以下过滤器是等效的，在Scala和Python中，结果是相同的:

df.filter(col("count") < 2).show(2)
df.where("count < 2").show(2)
-- in SQL
SELECT * FROM dfTable WHERE count < 2 LIMIT 2

输出为：

您可能希望将多个过滤器放入相同的表达式中。尽管这是可能的，但它并不总是有用的，因为Spark会自动执行所有的过滤操作，而不考虑过滤器的排序。这意味着，如果您想指定多个和过滤器，只需将它们按顺序链接起来，让Spark处理其余部分:

// in Scala
df.where(col("count") < 2).where(col("ORIGIN_COUNTRY_NAME") =!= "Croatia")
.show(2)

-- in SQL
SELECT * FROM dfTable WHERE count < 2 AND ORIGIN_COUNTRY_NAME != "Croatia"
LIMIT 2

输出为：

5.4.11.   获取唯一行

一个非常常见的用例是在一个DataFrame中提取惟一的或不同的值。这些值可以在一个或多个列中。我们这样做的方法是在DataFrame上使用不同的方法，它允许我们对该DataFrame中的任何行进行删除重复行。例如，让我们在数据集中获取唯一的起源地。当然，这是一个转换，它将返回一个新的DataFrame，只有唯一的行：

// in Scala
df.select("ORIGIN_COUNTRY_NAME", "DEST_COUNTRY_NAME").distinct().count()
-- in SQL
SELECT COUNT(DISTINCT(ORIGIN_COUNTRY_NAME, DEST_COUNTRY_NAME)) FROM dfTable

结果为256.

// in Scala
df.select("ORIGIN_COUNTRY_NAME").distinct().count()
# in Python
df.select("ORIGIN_COUNTRY_NAME").distinct().count()
-- in SQL
SELECT COUNT(DISTINCT ORIGIN_COUNTRY_NAME) FROM dfTable

结果为125。

5.4.12.   随机样本

有时，您可能只想从DataFrame中抽取一些随机记录。

您可以使用DataFrame上的sample方法来实现这一点，这使您可以指定要从DataFrame中提取指定比例的数据行，以及您是否希望使用或不使用替换:

val seed = 5
val withReplacement = false
val fraction = 0.5
df.sample(withReplacement, fraction, seed).count()

5.4.13.   随机分割

当您需要将您的DataFrame分割为原始DataFrame的随机“分割”时，随机分割将非常有用。它经常与机器学习算法一起用于创建训练、验证和测试集。在下一个示例中，我们将把DataFrame分成两种不同的DataFrame，通过设置权重来划分DataFrame(这些是函数的参数)，因为这个方法被设计成是随机的，所以我们还将指定一个种子(只需在代码块中以您所选择的数量替换种子)。需要注意的是，如果您没有为每个加起来为1的DataFrame指定一个比例，那么它们将被规范化，以便:

// in Scala
val dataFrames = df.randomSplit(Array(0.25, 0.75), seed)
dataFrames(0).count() > dataFrames(1).count() // False

5.4.14.   连接和附加行(union)

正如您在前一节中了解到的，DataFrames是不可变的。这意味着用户不能向DataFrames追加，因为这会改变它。要附加到DataFrame，必须将原始的DataFrame与新的DataFrame结合起来。这只是连接了两个DataFramess。对于union two DataFrames，您必须确保它们具有相同的模式和列数;否则，union将会失败。

注意
union目前是基于位置而不是模式执行的。这意味着列不会自动按照您认为的方式排列。

// in Scala
import org.apache.spark.sql.Row
val schema = df.schema
val newRows = Seq(
 Row("New Country", "Other Country", 5L),
 Row("New Country 2", "Other Country 3", 1L)
)
val parallelizedRows = spark.sparkContext.parallelize(newRows)
val newDF = spark.createDataFrame(parallelizedRows, schema)
df.union(newDF)
.where("count = 1")
.where($"ORIGIN_COUNTRY_NAME" =!= "United States")
.show() // get all of them and we'll see our new rows at the end

在Scala中，必须使用=!=运算符，这样您不仅可以将未求值的列表达式与字符串进行比较，还可以将其与已求值的列表达式进行比较。

输出为：

如预期的那样，您将需要使用这个新的DataFrame引用，以便引用带有新添加行的DataFrame。一种常见的方法是将DataFrame设置为视图，或者将其注册为表，以便在代码中更动态地引用它。

5.4.15.   行排序

在对DataFrame中的值进行排序时，我们总是希望对DataFrame顶部的最大或最小值进行排序。有两个相同的操作可以执行这种操作：sort和orderBy。它们接受列表达式和字符串以及多个列。默认是按升序排序:

// in Scala
df.sort("count").show(5)
df.orderBy("count", "DEST_COUNTRY_NAME").show(5)
df.orderBy(col("count"), col("DEST_COUNTRY_NAME")).show(5)

要更明确地指定排序方向，需要在操作列时使用asc和desc函数。这些允许您指定给定列的排序顺序:

// in Scala
import org.apache.spark.sql.functions.{desc, asc}
df.orderBy(expr("count desc")).show(2)
df.orderBy(desc("count"), asc("DEST_COUNTRY_NAME")).show(2)

一个高级技巧是使用asc_nulls_first、desc_nulls_first、asc_nulls_last或desc_nulls_last，来指定您希望在有序的DataFrame中显示null值的位置。

出于优化目的，有时建议在另一组转换之前对每个分区进行排序。您可以使用sortWithinPartitions方法来执行以下操作:

// in Scala
spark.read.format("json").load("/data/flight-data/json/*-summary.json")
.sortWithinPartitions("count")

我们将在第3部分中讨论调优和优化。

5.4.16.   limit

通常，您可能想要限制从DataFrame中提取的内容;例如，您可能只想要一些DataFrame的前十位。你可以使用极限法：

// in Scala
df.limit(5).show()
# in Python
df.limit(5).show()
-- in SQL
SELECT * FROM dfTable LIMIT 6
// in Scala
df.orderBy(expr("count desc")).limit(6).show()
# in Python
df.orderBy(expr("count desc")).limit(6).show()
-- in SQL
SELECT * FROM dfTable ORDER BY count desc LIMIT 6

5.4.17.   Repartition和Coalesce(重新分区和合并)

另一个重要的优化方式是根据一些经常过滤的列对数据进行分区，它控制跨集群的数据的物理布局，包括分区计划和分区数量。

Repartition将导致数据的完全shuffle，无论是否需要重新shuffle。这意味着只有当将来的分区数目大于当前的分区数目时，或者当您希望通过一组列进行分区时，您才应该使用Repartition:

// in Scala
df.rdd.getNumPartitions // 1

// in Scala
df.repartition(5)

如果经常对某个列进行过滤，那么基于该列进行重新分区是值得的:

// in Scala
df.repartition(col("DEST_COUNTRY_NAME"))

您也可以选择指定想要的分区数量:

// in Scala
df.repartition(5, col("DEST_COUNTRY_NAME"))

另一方面，Coalesce不会导致完全shuffle，并尝试合并分区。

这个操作将根据目标国家的名称将您的数据转移到5个分区中，然后合并它们(没有完全shuffle)：

// in Scala
df.repartition(5, col("DEST_COUNTRY_NAME")).coalesce(2)

5.4.18.   收集row到driver端

如前所述，Spark在驱动程序中维护集群的状态。有时，您需要收集一些数据给驱动程序，以便在本地机器上操作它。

到目前为止，我们还没有明确地定义这个操作。然而，我们使用了几种不同的方法来实现这一点，它们实际上都是相同的。

• collect从整个DataFrame中获取所有数据

• take选取DataFrame的前几行

• show打印出几行数据

// in Scala
val collectDF = df.limit(10)
collectDF.take(5) // take works with an Integer count
collectDF.show() // this prints it out nicely
collectDF.show(5, false)
collectDF.collect()

还有一种方法可以收集行到驱动程序，以便迭代整个数据集。方法toLocalIterator将把分区收集到驱动程序作为迭代器。此方法允许以分区为单位方式遍历整个数据集：

collectDF.toLocalIterator()

注意
任何数据收集到driver端，都可能是一个非常昂贵的操作! 如果您有一个大的数据集，调用collect，可能使驱动程序崩溃。如果您使用toLocalIterator，并且有非常大的分区，也可能很容易地崩溃driver节点并丢失应用程序的状态。

5.5.  总结

本章介绍了DataFrame的基本操作。学习了使用Spark DataFrames所需的简单概念和工具。第6章更详细地介绍DataFrame中操作数据的所有不同方法。