【技术分享】Spark DataFrame入门手册

本文原作者：赖博先，经授权后发布。

一、简介

Spark SQL是spark主要组成模块之一，其主要作用与结构化数据，与hadoop生态中的hive是对标的。而DataFrame是spark SQL的一种编程抽象，提供更加便捷同时类同与SQL查询语句的API，让熟悉hive的数据分析工程师能够非常快速上手。

DataFrame是一种以命名列的方式组织的分布式数据集，可以类比于hive中的表。但是比hive表更加灵活的是，你可以使用各种数据源来构建一个DataFrame，如：结构化数据文件（例如json数据）、hive表格、外部数据库，还可以直接从已有的RDD变换得来。后面会把相关方法、接口跟大家一一道来。

二、初步使用

大家学习一门语言可能都是从“hello word！”开始的，这主要目的是让学习者熟悉程序运行的环境，同时亲身感受程序运行过程。这里我们也会从环境到运行的步骤进行讲解。

导入spark运行环境相关的类

所有spark相关的操作都是以sparkContext类作为入口，而Spark SQL相关的所有功能都是以SQLContext类作为入口。下面的语句是新建入口类的对象。最下面的语句是引入隐式转换，隐式的将RDD转换为DataFrame。

下面就是从tdw表中读取对应的表格数据，然后就可以使用DataFrame的API来操作数据表格，其中TDWSQLProvider是数平提供的spark tookit，可以在KM上找到这些API的详细用法。

这段代码的意思是从tdw 表中读取对应分区的数据，select出表格中对应的字段（这里面的字段名字就是表格字段名字，需要用双引号）toDF将筛选出来的字段转换成DataFrame，在进行groupBy操作，这里的groupBy操作跟TDW hive操作是一样的意思，对指定字段进行分组操作，count函数用来计数计数，这里得到的DataFrame最后有一个”count”命名的字段保存每个分组的个数（这里特别需要注意函数的返回类型，groupby函数返回的并不是dataframe类型的数据，后面会提到）。接下来的printSchema函数是打印出edge的视图，可以理解成tdw idea里面的show DDL，Show函数是打印出这个DataFrame前20行数据（默认），当然可以指定行数打印。

从上面的例子中可以看出，DataFrame基本把SQL函数给实现了，在hive中用到的很多操作（如：select、groupBy、count、join等等）可以使用同样的编程习惯写出spark程序，这对于没有函数式编程经验的同学来说绝对福利。

三、函数说明及其用法

函数式编程是spark编程的最大特点，而函数则是函数式编程的最小操作单元，这边主要列举DataFrame常用函数以及主要用法：

Action 操作

特别注意每个函数的返回类型

1、 collect() ,返回值是一个数组，返回dataframe集合所有的行

2、 collectAsList() 返回值是一个java类型的数组，返回dataframe集合所有的行

3、 count() 返回一个number类型的，返回dataframe集合的行数

4、 describe(cols: String*) 返回一个通过数学计算的类表值(count, mean, stddev, min, and max)，这个可以传多个参数，中间用逗号分隔，如果有字段为空，那么不参与运算，只这对数值类型的字段。例如df.describe("age", "height").show()

5、 first() 返回第一行，类型是row类型

6、 head() 返回第一行，类型是row类型

7、 head(n:Int)返回n行，类型是row 类型

8、 show()返回dataframe集合的值默认是20行，返回类型是unit

9、 show(n:Int)返回n行，，返回值类型是unit

10、 table(n:Int) 返回n行，类型是row 类型

DataFrame的基本操作

1、 cache()同步数据的内存

2、 columns 返回一个string类型的数组，返回值是所有列的名字

3、 dtypes返回一个string类型的二维数组，返回值是所有列的名字以及类型

4、 explan()打印执行计划

5、 explain(n:Boolean) 输入值为 false 或者true ，返回值是unit 默认是false ，如果输入true 将会打印逻辑的和物理的

6、 isLocal 返回值是Boolean类型，如果允许模式是local返回true 否则返回false

7、 persist(newlevel:StorageLevel) 返回一个dataframe.this.type 输入存储模型类型

8、 printSchema() 打印出字段名称和类型按照树状结构来打印

9、 registerTempTable(tablename:String) 返回Unit ，将df的对象只放在一张表里面，这个表随着对象的删除而删除了

10、 schema 返回structType 类型，将字段名称和类型按照结构体类型返回

11、 toDF()返回一个新的dataframe类型的

12、 toDF(colnames：String*)将参数中的几个字段返回一个新的dataframe类型的，

13、 unpersist() 返回dataframe.this.type 类型，去除模式中的数据

14、 unpersist(blocking:Boolean)返回dataframe.this.type类型 true 和unpersist是一样的作用false 是去除RDD

聚合函数：

1、 agg(expers:column*) 返回dataframe类型，同数学计算求值

df.agg(max("age"), avg("salary"))

df.groupBy().agg(max("age"), avg("salary"))

2、 agg(exprs: Map[String, String]) 返回dataframe类型，同数学计算求值 map类型的

df.agg(Map("age" -> "max", "salary" -> "avg"))

df.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))

3、 agg(aggExpr: (String, String), aggExprs: (String, String)*) 返回dataframe类型，同数学计算求值

df.agg(Map("age" -> "max", "salary" -> "avg"))

df.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))

4、 apply(colName: String) 返回column类型，捕获输入进去列的对象

5、 as(alias: String) 返回一个新的dataframe类型，就是原来的一个别名

6、 col(colName: String) 返回column类型，捕获输入进去列的对象

7、 cube(col1: String, cols: String*) 返回一个GroupedData类型，根据某些字段来汇总

8、 distinct 去重返回一个dataframe类型

9、 drop(col: Column) 删除某列返回dataframe类型

10、 dropDuplicates(colNames: Array[String]) 删除相同的列返回一个dataframe

11、 except(other: DataFrame) 返回一个dataframe，返回在当前集合存在的在其他集合不存在的；这个操作非常有用呀

12、 explode[A, B](inputColumn: String, outputColumn: String)(f: (A) ⇒ TraversableOnce[B])(implicit arg0: scala.reflect.api.JavaUniverse.TypeTag[B]) 返回值是dataframe类型，这个将一个字段进行更多行的拆分

df.explode("name","names") {name :String=> name.split(" ")}.show();

将name字段根据空格来拆分，拆分的字段放在names里面

13、 filter(conditionExpr: String): 刷选部分数据，返回dataframe类型

df.filter("age>10").show();

df.filter(df("age")>10).show();

df.where(df("age")>10).show(); 都可以

14、 groupBy(col1: String, cols: String*) 根据某写字段来汇总返回groupedate类型

df.groupBy("age").agg(Map("age"->"count")).show();

df.groupBy("age").avg().show();都可以

这里如果要把groupBy之后的结果转换成一个Dataframe需要另一个函数转换一下，比如 count

15、 intersect(other: DataFrame) 返回一个dataframe，在2个dataframe都存在的元素

16、 join(right: DataFrame, joinExprs: Column, joinType: String)

一个是关联的dataframe，第二个关联的条件，第三个关联的类型：inner, outer, left_outer, right_outer, leftsemi

df.join(ds,df("name")===ds("name")

and df("age")===ds("age"),"outer").show();

17、 limit(n: Int) 返回dataframe类型去n 条数据出来

18、 na: DataFrameNaFunctions ，可以调用dataframenafunctions的功能区做过滤df.na.drop().show(); 删除为空的行

19、 orderBy(sortExprs: Column*) 做alise排序，还可以指定进行降序排序desc

20、 select(cols:string*) dataframe 做字段的刷选

df.select($"colA", $"colB" 1)

这里面select有两种类型的参数，一种是上面的string类型，就是前面没有$符号，如果加了$标识这是一个column类型。使用这种类型需要加import sqlContext.implicits._ （这些是从身边spark大神xuehao同学那里学到的）这些细节真的从实践中来，所以大家赶紧收藏！

21、selectExpr(exprs: String*) 做字段的刷选

df.selectExpr("name","name as names","upper(name)","age 1").show();

22、 sort(sortExprs: Column*) 排序

df.sort(df("age").desc).show(); 默认是asc；这里的写法可以有很多几种。

23、 unionAll(other:Dataframe) 合并 df.unionAll(ds).show();

24、 withColumnRenamed(existingName: String, newName: String)

修改列表 df.withColumnRenamed("name","names").show();

25、 withColumn(colName: String, col: Column) 增加一列

df.withColumn("aa",df("name")).show();