DataFrame由记录（record）组成，record是Row类型（与一个table中的行相似）。一条record由多列组成（类似于一个电子表格中的列），列表示可以在该Dataset中每个单独的记录上执行的计算表达式。模式定义了DataFrame列的名以及列的数据类型。DataFrame的分区定义了DataFrame以及Dataset在集群上的物理分布，而划分模式定义了partition的分配方式，你可以自定义分区的方式，也可以采取随机分配的方式

DataFrame有很多列，而模式定义了这些列的名字和数据类型。可以用下面的方法查询模式

df.printSchema()

模式

模式定义DataFrame的列名以及列的数据类型，它可以由数据源来定义模式（称为读时模式，schema-on-read），也可以由我们自己来显式地定义。

实际应用场景决定了定义Schema的方式，当应用于即席分析时，使用读时模式即可。但是，这也可能导致数据精度损失问题，例如在文件中读取时，将long型错误地解析为整数。当使用Spark进行实现生产级别ETL的时候，最好采取显式定义Schema的方式，尤其是在处理诸如CSV和JSON之类的无类型数据源时更是如此，这是因为模式推断方法会根据读入数据类型而变化

一个模式是由许多字段构成的StructType，这些字段即为StructField，具有名称、类型、布尔标志（该标志指定该列是否可以包含缺失值或空值），并且用户可指定与该列关联的元数据。元数据存储着有关此列的信息。

以下示例演示了如何为一个DataFrame创建并指定模式。

importorg.apache.spark.sql.types.{StructField,StructType,StringType,LongType}importorg.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\"}"))))

注意：
Spark 在读取 JSON 文件的时候，会忽略你手动设置的nullable=false，强制把字段的可空属性改为 true。
JSON 文件里，某一行可以缺失字段、字段值为 null。Spark 出于安全层面考虑：原始数据源无法保证字段一定存在，就算你代码里写了false，Spark 仍然会强制将该字段的 nullable 设置为true。

列和表达式

Spark中的列与电子表格，R dataframe或pandas DataFrame中的列类似，可以对DataFrame中的列进行选择、转换操作和删除，并将这些操作表示为表达式

对于Spark而言，列是逻辑结构，它只是表示根据表达式为每个记录计算出的值。这意味着要为一列创建一个真值，我们需要有一行，而要有一行则需要有一个DataFrame，不能在DataFrame的范围外操作一个列，必须对DataFrame使用Spark的转换操作来修改列的内容

列

有很多不同的方法来构造和引用列，两个最简单的方法是通过col函数或column函数。使用这两个函数，需要传入列名

在Scala中有一些特有的语言支持，可以使用更多简短的方式来引用列

显式列引用

如果需要引用某DataFrame的某一列，则可以在这个DataFrame上使用col方法，当执行连接操作时，如果两个连接的DataFrame存在一个同名列，该方法会非常有用，显式引用列的另一个好处就是Spark不用自己解析该列（在分析阶段）：

df.col("count")

表达式

表达式是对一个DataFrame中某一个记录的一个或多个值的一组转换操作，把他想象成一个函数，它将一个或多个列名作为输入，解析它们，然后针对数据集中的每条记录应用表达式来得到一个单值，这个“单值”实际上可以是一个复杂的类型，如Map或Array。

在最简单的情况下，通过expr函数创建的表达式，仅仅是一个DataFrame列的引用，也就是说，expr(“somCol”)等同于col(“someCol”)

列作为表达式

列提供了表达式功能的一个子集。如果你使用col()，并想对该列执行转换操作，则必须对该列的引用执行这些转换操作。当使用表达式时，expr函数实际上可以将字符串解析成转换操作和列引用，也可以在之后将其传递到下一步的转换操作

访问DataFrame的列

有时，可以使用printSchema查询DataFrame的列，但是要想在程序中访问列，可以使用属性columns查询DataFrame的所有列：

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

记录和行

在Spark中，DataFrame的每一行都是一个记录，而记录是Row类型的对象。Spark使用列表达式操纵Row类型对象。Row对象内部其实是字节数组，但是Spark没有提供访问这些数组的接口，因此我们只能使用列表达式去操纵。

当使用DataFrame时，向驱动器请求行的命令总是返回一个或多个Row类型的行数据

我们可以通过在DataFrame上调用first()来查看一行：

创建行

可以基于已知的每列数值去手动实例化一个Row对象来创建行，需要注意的是，只有DataFrame具有模式，行对象本身没有模式，这意味着，如果手动创建Row对象，则必须按照该行所附属的Dataframe的列顺序来初始化Row对象

importorg.apache.spark.sql.Row val myRow=Row("hello",null,1,false)

访问行的数据也同样简单：只需指定你想要的位置。使用Scala或Java时，你必须使用辅助方法或显式地指定值类型

DataFrame转换操作

创建DataFrame

我们可以原始数据源中创建DataFrame

也可以通过获取一组行并将它们转换操作为一个DataFrame来即时创建DataFrame

importorg.apache.spark.sql.Rowimportorg.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中，还可以通过在Seq类型上运行toDF函数来利用控制台中的Spark隐式方法（如果已经将它们导入JAR代码中），由于对于null类型的支持并不稳定，所以并不推荐在实际生产中使用。

val myDF=Seq(("Hello",3,1L)).toDF("col1","col2","col3")

接下来看看DataFrame类型支持的最有用的方法：处理列或表达式时的select方法，以及处理字符串表达式时的selectExpr方法

当然有的转换操作不是针对列的操作方法，因此org.apache.spark.sql.functions包中包含一组函数方法用来额外支持

select函数和selectExpr函数

select函数和selectExpr函数支持在DataFrame上执行类似数据表的SQL查询

简单来说，可以使用select和selectExpr来操作DataFrame中的列。最简单的方式就是使用select方法，待处理的列名字符串作为参数传递

可以使用相同格式的查询来选择多个列，只需在select方法调用中添加更多的列名字符串即可

可以通过多种不同的方式引用列，而且这些方式可以等价互换：

importorg.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

一个常见的错误是混淆Column对象和字符串。例如，以下代码将导致编译错误：

expr是我们目前使用到的最灵活的引用方式。它能够引用一列，也可以引用对列进行操纵的字符串表达式。

因为select后跟着一系列expr是非常常见的写法，所以Spark有一个有效地描述此操作序列的接口：selectExpr，它可能是最常用的接口：

df.selectExpr("dest_country_name as destination","dest_country_name").show

我们可以利用selectExpr构建复杂表达式来创建DataFrame，实际上，我们可以添加任何不包含聚合操作的有效SQL语句，并且只要列可以解析，它就是有效的！
下面是一个简单的例子，在DataFrame中增加一个新列withinCountry，该列描述了destination和origin是否相同

df.selectExpr("*","(dest_country_name = origin_country_name) as withinCountry").show

使用select语句，我们还可以利用系统预定义好的聚合函数来指定在整个DataFrame上的聚合操作

df.selectExpr("avg(count)","count(distinct(dest_country_name))").show

转换操作成Spark类型（字面量）

有时候需要给Spark传递显式的值，它们只是一个值而非新列。这可能是一个常量值，或接下来需要比较的值。我们的方式是通过字面量（literal）传递，简单来说，就是将给定编程语言的字面上的值转换操作为Spark可以理解的值。字面量就是表达式，可以用操作表达式的方式来使用它们：

importorg.apache.spark.sql.functions.lit df.select(expr("*"),lit(1).as("one")).show

添加列

使用withColumn可以为DataFrame增加新列，这种方式更为规范一些

df.withColumn("numberOne",lit(1)).show

下个示例中，当出发国家与目的地国家相同时，我们将为其设置一个布尔标志：

注意，withColumn函数有两个参数：列名和为给定行赋值的表达式。

也可以用withColumn对某一列重命名

重命名列

我们不仅可以使用withColumn对列重命名，还可以使用withColumnRenamed方法实现对列重命名
第一个参数是要被修改的列的名，第二个参数是新的列名

保留字与关键字

你可能会遇到列名中包含空格或者连字符等保留字符，要处理这些保留字符意味着要适当地对列名进行转义，在Spark中，我们通过反引号字符来实现。withColumn允许使用保留字来创建列

这里不需要转义字符，因为withColumn的第一个参数只是新列名的字符串
但在下面例子中，我们需要使用反引号，因为我们在表达式中引用了一个列

如果我们显式地使用字符串来引用列，则可以引用带有保留字符的类（而不用转义他们），这个字符串会被解释成字面值，而不是表达式。我们只需要转义使用保留字符或者关键字的表达式

区分大小写

Spark默认是不区分大小写的，但是可以通过如下配置使Spark区分大小写

spark.conf.set("spark.sql.caseSensitive","true")

删除列

我们可以通过select实现，也可以使用drop方法来删除列

可以通过传入多个列作为参数来实现删除多个列

更改列的类型（强制类型转换）

有时，我们可能需要将某一列类型转换操作为另一种类型

df.withColumn("count2",col("count").cast("int")).printSchema

过滤行

为了过滤行，只要创建一个表达式来判断该表达式是true还是false，然后过滤使表达式为false的行。在DataFrame上实现过滤操作最常见的方法是创建一个字符串表达式，或者通过列操作来构建表达式。有两种实现过滤的方式，分别是where和filter，他们可以执行相同的操作，接受相同参数类型。

我们可能本能地想把多个过滤条件放到一个表达式中，尽管这种方式可行，但是并不总有效，因为Spark会同时执行所有过滤操作，不管过滤条件的先后顺序，因此当你想要指定多个AND过滤操作时，只要按照先后顺序以链式的方式把这些过滤条件串联起来，然后让Spark执行剩下的工作

df.where(col("count")<2).where(col("origin_country_name")=!="Croatia").show

获得去重后的行

一个常见的应用场景是去除DataFrame中重复的行，这可以去除一列或者多列中重复的值，我们实现去重的方式是使用DataFrame的distinct方法，它能够对DataFrame中的行进行去重操作。例如，从数据集中找出所有不同的出发国家-目的地国家组合或者所有不同的出发国家

随机抽样

有时可能想从DataFrame中随机抽取一些记录，可以使用DataFrame的sample方法来实现此操作，它按一定比例从DataFrame中随机抽取一部分行，也可以通过withReplacement参数指定是否放回抽样，true为有放回的抽样（可以有重复样本），false为无放回的抽样

连接和追加行

DataFrame是不可变的，这意味着用户不能向DataFrame追加行，如果想要向DataFrame追加行，必须将原始DataFrame与新的DataFrame联合起来，即union操作，也就是拼接两个DataFrame。若想联合两个DataFrame，你必须确保它们具有相同的模式和列数，否则联合操作将会失败

目前，联合操作是基于位置而不是基于数据模式schema执行，也就是说，它并不会自动根据列名匹配对齐后再进行联合，所以两个联合的DataFrame需要具有完全相同的模式和列数

importorg.apache.spark.sql.Row val schema=df.schema val newRows=Seq(Row("New Country","Other Country",5L), Row("New Country","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

当DataFrame追加了记录后，需要对产生的新DataFrame进行引用，一个常见的方式是将这个新DataFrame变成视图（View）或者注册成一个数据表，以便在代码中使用

行排序

sort和orderBy方法是相互等价的操作，执行的方式也一样。它们均接收列表达式和字符串，以及多个列。默认设置是按升序排序：

若要明确地指定升序还是降序，则需使用asc函数和desc函数

df.sort(desc("origin_country_name"),asc("count")).show

一个高级技巧是你可以指定空值在排序列表中的位置，使用asc_nulls_first指示空值安排在升序排列的前面，desc_nulls_first指示空值安排在降序排序的前面，asc_nulls_last指示空值安排在升序排序的后面，desc_nulls_last指示空值安排在降序排序的后面

出于性能优化的目的，最好是在进行别的转换之前，先对每个分区进行内部排序。可以使用sortWithinPartitions方法实现这一操作

limit方法

通常，你可能想限制从DataFrame中提取的内容。可以使用limit实现：

重划分和合并

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

不管是否有必要，重新分区都会导致数据的全面洗牌，如果将来的分区数大于当前的分区数，或者当你想要基于某一组特定列来进行分区时，通常只能重新分区

如果你知道你经常按某一列执行过滤操作，则根据该列进行重新分区是很有必要的

你还可以指定你想要的分区数量

而另一方面，合并（coalesce）不会导致数据的全面洗牌，但会尝试合并分区，下面的示例代码将基于目的地国家名的列将数据重新划分成5个分区，然后再合并它们（没有导致数据全面洗牌）

驱动器获取行

Spark的驱动器维护者集群状态，有时候你需要让驱动器收集一些数据到本地，这样你可以在本地机器上处理它们。

到目前为止，我们并没有明确定义这个操作，但我们使用了几种不同的方法来实现完全相同的效果，下面的代码示例使用collect函数从整个DataFrame中获取所有数据，使用take函数选择前N行，并使用show函数打印一些行

为了遍历整个数据集，还有一种让驱动器获取行的方法，即toLocalIterator函数。toLocalIterator函数是一个迭代器，将每个分区的数据返回给驱动器。这个函数允许你以串行的方式一个一个分区地迭代整个数据集