Spark SQL中的DataFrame类似于一张关系型数据表。在关系型数据库中对单表或进行的查询操作，在DataFrame中都可以通过调用其API接口来实现。可以参考，Scala提供的DataFrame API。

本文中的代码基于Spark-1.6.2的文档实现。

一、DataFrame对象的生成

Spark-SQL可以以其他RDD对象、parquet文件、json文件、Hive表，以及通过JDBC连接到其他关系型数据库作为数据源来生成DataFrame对象。本文将以MySQL数据库为数据源，生成DataFrame对象后进行相关的DataFame之上的操作。

文中生成DataFrame的代码如下：

object DataFrameOperations {

def main (args: Array[String ]) {

val sparkConf = new SparkConf().setAppName( "Spark SQL DataFrame Operations").setMaster( "local[2]" )

val sparkContext = new SparkContext(sparkConf)

val sqlContext = new SQLContext(sparkContext)

val url = "jdbc:mysql://m000:3306/test"

val jdbcDF = sqlContext.read.format( "jdbc" ).options(

Map( "url" -> url,

"user" -> "root",

"password" -> "root",

"dbtable" -> "spark_sql_test" )).load()

val joinDF1 = sqlContext.read.format( "jdbc" ).options(

Map("url" -> url ,

"user" -> "root",

"password" -> "root",

"dbtable" -> "spark_sql_join1" )).load()

val joinDF2 = sqlContext.read.format( "jdbc" ).options(

Map ( "url" -> url ,

"user" -> "root",

"password" -> "root",

"dbtable" -> "spark_sql_join2" )).load()

... ...

}

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

后续代码都在上面... ...处。

二、DataFrame对象上Action操作

1、show：展示数据

以表格的形式在输出中展示jdbcDF中的数据，类似于select * from spark_sql_test的功能。

show方法有四种调用方式，分别为，

(1)show

只显示前20条记录。

示例：

jdbcDF.show

1

1

结果：

(2)show(numRows: Int)

显示numRows条

示例：

jdbcDF.show(3)

1

1

结果：

(3)show(truncate: Boolean)

是否最多只显示20个字符，默认为true。

示例：

jdbcDF.show(true)

jdbcDF.show(false)

1

2

1

2

结果：

(4)show(numRows: Int, truncate: Boolean)

综合前面的显示记录条数，以及对过长字符串的显示格式。

示例：

jdbcDF.show(3, false)

1

1

结果：

2、collect：获取所有数据到数组

不同于前面的show方法，这里的collect方法会将jdbcDF中的所有数据都获取到，并返回一个Array对象。

jdbcDF.collect()

1

1

结果如下，结果数组包含了jdbcDF的每一条记录，每一条记录由一个GenericRowWithSchema对象来表示，可以存储字段名及字段值。

3、collectAsList：获取所有数据到List

功能和collect类似，只不过将返回结构变成了List对象，使用方法如下

jdbcDF.collectAsList()

1

1

结果如下，

4、describe(cols: String*)：获取指定字段的统计信息

这个方法可以动态的传入一个或多个String类型的字段名，结果仍然为DataFrame对象，用于统计数值类型字段的统计值，比如count, mean, stddev, min, max等。

使用方法如下，其中c1字段为字符类型，c2字段为整型，c4字段为浮点型

jdbcDF .describe("c1" , "c2", "c4" ).show()

1

1

结果如下，

5、first, head, take, takeAsList：获取若干行记录

这里列出的四个方法比较类似，其中

(1)first获取第一行记录

(2)head获取第一行记录，head(n: Int)获取前n行记录

(3)take(n: Int)获取前n行数据

(4)takeAsList(n: Int)获取前n行数据，并以List的形式展现

以Row或者Array[Row]的形式返回一行或多行数据。first和head功能相同。

take和takeAsList方法会将获得到的数据返回到Driver端，所以，使用这两个方法时需要注意数据量，以免Driver发生OutOfMemoryError

使用和结果略。

二、DataFrame对象上的条件查询和join等操作

以下返回为DataFrame类型的方法，可以连续调用。

1、where条件相关

(1)where(conditionExpr: String)：SQL语言中where关键字后的条件

传入筛选条件表达式，可以用and和or。得到DataFrame类型的返回结果，

示例：

jdbcDF .where("id = 1 or c1 = 'b'" ).show()

1

1

结果，

(2)filter：根据字段进行筛选

传入筛选条件表达式，得到DataFrame类型的返回结果。和where使用条件相同

示例：

jdbcDF .filter("id = 1 or c1 = 'b'" ).show()

1

1

结果，

2、查询指定字段

(1)select：获取指定字段值

根据传入的String类型字段名，获取指定字段的值，以DataFrame类型返回

示例：

jdbcDF.select( "id" , "c3" ).show( false)

1

1

结果：

还有一个重载的select方法，不是传入String类型参数，而是传入Column类型参数。可以实现select id, id+1 from test这种逻辑。

jdbcDF.select(jdbcDF( "id" ), jdbcDF( "id") + 1 ).show( false)

1

1

结果：

能得到Column类型的方法是apply以及col方法，一般用apply方法更简便。

(2)selectExpr：可以对指定字段进行特殊处理

可以直接对指定字段调用UDF函数，或者指定别名等。传入String类型参数，得到DataFrame对象。

示例，查询id字段，c3字段取别名time，c4字段四舍五入：

jdbcDF .selectExpr("id" , "c3 as time" , "round(c4)" ).show(false)

1

1

结果，

(3)col：获取指定字段

只能获取一个字段，返回对象为Column类型。

val idCol = jdbcDF.col(“id”)果略。

(4)apply：获取指定字段

只能获取一个字段，返回对象为Column类型

示例：

val idCol1 = jdbcDF.apply("id")

val idCol2 = jdbcDF("id")

1

2

1

2

结果略。

(5)drop：去除指定字段，保留其他字段

返回一个新的DataFrame对象，其中不包含去除的字段，一次只能去除一个字段。

示例：

jdbcDF.drop("id")

jdbcDF.drop(jdbcDF("id"))

1

2

1

2

结果：

3、limit

limit方法获取指定DataFrame的前n行记录，得到一个新的DataFrame对象。和take与head不同的是，limit方法不是Action操作。

jdbcDF.limit(3).show( false)

1

1

结果，

4、order by

(1)orderBy和sort：按指定字段排序，默认为升序

示例1，按指定字段排序。加个-表示降序排序。sort和orderBy使用方法相同

jdbcDF.orderBy(- jdbcDF("c4")).show(false)

// 或者

jdbcDF.orderBy(jdbcDF("c4").desc).show(false)

1

2

3

1

2

3

结果，

示例2，按字段字符串升序排序

jdbcDF.orderBy("c4").show(false)

1

1

结果，

(2)sortWithinPartitions

和上面的sort方法功能类似，区别在于sortWithinPartitions方法返回的是按Partition排好序的DataFrame对象。

5、group by

(1)groupBy：根据字段进行group by操作

groupBy方法有两种调用方式，可以传入String类型的字段名，也可传入Column类型的对象。

使用方法如下，

jdbcDF .groupBy("c1" )

jdbcDF.groupBy( jdbcDF( "c1"))

1

2

1

2

(2)cube和rollup：group by的扩展

功能类似于SQL中的group by cube/rollup，略。

(3)GroupedData对象

该方法得到的是GroupedData类型对象，在GroupedData的API中提供了group by之后的操作，比如，

max(colNames: String*)方法，获取分组中指定字段或者所有的数字类型字段的最大值，只能作用于数字型字段

min(colNames: String*)方法，获取分组中指定字段或者所有的数字类型字段的最小值，只能作用于数字型字段

mean(colNames: String*)方法，获取分组中指定字段或者所有的数字类型字段的平均值，只能作用于数字型字段

sum(colNames: String*)方法，获取分组中指定字段或者所有的数字类型字段的和值，只能作用于数字型字段

count()方法，获取分组中的元素个数

运行结果示例：

count

max

这里面比较复杂的是以下两个方法，

agg，该方法和下面介绍的类似，可以用于对指定字段进行聚合操作。

pivot

6、distinct

(1)distinct：返回一个不包含重复记录的DataFrame

返回当前DataFrame中不重复的Row记录。该方法和接下来的dropDuplicates()方法不传入指定字段时的结果相同。

示例：

jdbcDF.distinct()

1

1

结果，

(2)dropDuplicates：根据指定字段去重

根据指定字段去重。类似于select distinct a, b操作

示例：

jdbcDF.dropDuplicates(Seq("c1"))

1

1

结果：

7、聚合

聚合操作调用的是agg方法，该方法有多种调用方式。一般与groupBy方法配合使用。

以下示例其中最简单直观的一种用法，对id字段求最大值，对c4字段求和。

jdbcDF.agg("id" -> "max", "c4" -> "sum")

1

1

结果：

8、union

unionAll方法：对两个DataFrame进行组合

类似于SQL中的UNION ALL操作。

示例：

jdbcDF.unionALL(jdbcDF.limit(1))

1

1

结果：

9、join

重点来了。在SQL语言中用得很多的就是join操作，DataFrame中同样也提供了join的功能。

接下来隆重介绍join方法。在DataFrame中提供了六个重载的join方法。

(1)、笛卡尔积

joinDF1.join(joinDF2)

1

1

(2)、using一个字段形式

下面这种join类似于a join b using column1的形式，需要两个DataFrame中有相同的一个列名，

joinDF1.join(joinDF2, "id")

1

1

joinDF1和joinDF2根据字段id进行join操作，结果如下，using字段只显示一次。

(3)、using多个字段形式

除了上面这种using一个字段的情况外，还可以using多个字段，如下

joinDF1.join(joinDF2, Seq("id", "name"))

1

1

(4)、指定join类型

两个DataFrame的join操作有inner, outer, left_outer, right_outer, leftsemi类型。在上面的using多个字段的join情况下，可以写第三个String类型参数，指定join的类型，如下所示

joinDF1.join(joinDF2, Seq("id", "name"), "inner")

1

1

(5)、使用Column类型来join

如果不用using模式，灵活指定join字段的话，可以使用如下形式

joinDF1.join(joinDF2 , joinDF1("id" ) === joinDF2( "t1_id"))

1

1

结果如下，

(6)、在指定join字段同时指定join类型

如下所示

joinDF1.join(joinDF2 , joinDF1("id" ) === joinDF2( "t1_id"), "inner")

1

1

10、获取指定字段统计信息

stat方法可以用于计算指定字段或指定字段之间的统计信息，比如方差，协方差等。这个方法返回一个DataFramesStatFunctions类型对象。

下面代码演示根据c4字段，统计该字段值出现频率在30%以上的内容。在jdbcDF中字段c1的内容为"a, b, a, c, d, b"。其中a和b出现的频率为2 / 6，大于0.3

jdbcDF.stat.freqItems(Seq ("c1") , 0.3).show()

1

1

结果如下：

11、获取两个DataFrame中共有的记录

intersect方法可以计算出两个DataFrame中相同的记录，

jdbcDF.intersect(jdbcDF.limit(1)).show(false)

1

1

结果如下：

12、获取一个DataFrame中有另一个DataFrame中没有的记录

示例：

jdbcDF.except(jdbcDF.limit(1)).show(false)

1

1

结果如下，

13、操作字段名

(1)withColumnRenamed：重命名DataFrame中的指定字段名

如果指定的字段名不存在，不进行任何操作。下面示例中将jdbcDF中的id字段重命名为idx。

jdbcDF.withColumnRenamed( "id" , "idx" )

1

1

结果如下：

(2)withColumn：往当前DataFrame中新增一列

whtiColumn(colName: String , col: Column)方法根据指定colName往DataFrame中新增一列，如果colName已存在，则会覆盖当前列。

以下代码往jdbcDF中新增一个名为id2的列，

jdbcDF.withColumn("id2", jdbcDF("id")).show( false)

1

1

结果如下，

14、行转列

有时候需要根据某个字段内容进行分割，然后生成多行，这时可以使用explode方法

下面代码中，根据c3字段中的空格将字段内容进行分割，分割的内容存储在新的字段c3_中，如下所示

jdbcDF.explode( "c3" , "c3_" ){time: String => time.split( " " )}

1

1

结果如下，