DataFrames具有如下特点：

• Ability to scale from kilobytes of data on a single laptop to petabytes on a large cluster（支持单机KB级到集群PB级的数据处理）
• Support for a wide array of data formats and storage systems（支持多种数据格式和存储系统，如图所示）
• State-of-the-art optimization and code generation through the Spark SQL Catalyst optimizer（通过Spark SQL Catalyst优化器可以进行高效的代码生成和优化）
• Seamless integration with all big data tooling and infrastructure via Spark（能够无缝集成所有的大数据处理工具）
• APIs for Python, Java, Scala, and R (in development via SparkR)

生成一个DF进行测试

 import sparkSession.implicits._
 var test_df = Seq((1,Array("1.0")),(2,Array("2.0")),(3,Array("3.0"))).toDF("imei","feature")

1.选出若干列（字段）

1.1 选出一列(字段)

imei_features_df.col("id")
org.apache.spark.sql.Column
（col返回为Column,select 返回为dataframe，df(col_name)为Column类型）
或
imei_features_df.select((imei_features_df.col("value") + 10).as("value2")).show//修改了把一列的值加十，再修改列名
org.apache.spark.sql.DataFrame
或
imei_features_df.select("value").show
或
imei_features_df.selectExpr("value") 选出名字为value的一列。
查看两列（两个字段）
imei_features_df.select(col("name"),col("value")).show
df.select(df("name"),df("value")+1).show()
df.select("name","value").show() select操作，典型的弱类型，untyped操作
不能写成 df.select("name","value"+1).show()这样。
df.select($"name", $"value" + 1).show() // 使用表达式，scala的语法，要用$符号作为前缀
另外selectExpr：也可以对指定字段进行特殊处理：
df.selectExpr("id", "col2 as time", "round(col3)").show(false)
【show方法参数默认为true，如果字段太长则会略去，改为false，会显示整个字段】

用apply：获取指定字段（只能一个，返回为Column）

     val id1 = df.apply("id")
     val id2 = df("id")

1.2 选出两列，把其中一列加1

imei_features_df.select(col("name"),col("value") + 1).show
或
imei_features_df.select(imei_features_df("name"),imei_features_df("value") + 1).show

1.3 获取指定字段统计信息

stat方法可以用于计算指定字段或指定字段之间的统计信息，比如方差，协方差等。这个方法返回一个DataFramesStatFunctions类型对象。
df.stat.freqItems(Seq ("c1","c2")).show()
　　下面代码演示根据c4字段，统计c1字段值出现频率在30%以上的内容。在df中字段c1的内容为”a, b, a, c, d, b”。其中a和b出现的频率为2 / 6，大于0.3
df.stat.freqItems(Seq ("c1") , 0.3).show()

1.4 describe

导入数据后执行的第一个操作是了解它们的大致情况. 对于数字列, 了解描述性摘要统计信息对理解数据的分布有很大帮助. 可以使用describe函数来返回一个DataFrame, 其中会包含非空项目数, 平均值, 标准偏差以及每个数字列的最小值和最大值等信息.
df.describe().show()

2.选取若干行记录

2.1 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

2.2 limit

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

3.删除操作

3.1.删除指定字段（列），保留其他字段

df.drop("id")
df.drop(df("id"))

3.2.取出前n行记录，得到一个新的dataframe。take和head是Action操作，limit则不是。

df.limit(n)

3.3.删除空值

df.na.drop() 删除带有空值的行

4.排序

4.1 orderBy和sort: 按指定字段排序，默认为升序,用法相同

//降序排列，只对数字类型和日期类型生效
df.orderBy( - df("age")).show(false)//加"-"表示降序排列
df.orderBy(df("time").desc).show

4.2 sortWithinPartitions

和sort类似，但是是使用Partition来对其他字段排序

4.group by

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

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

df.groupBy("col1")
df.groupBy(df("col"))

//分组计数
df.groupBy("age").count().show()

(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()方法，获取分组中的元素个数
agg方法，可以对指定字段进行聚合操作。

5. 去重

（1）distinct:返回一个不包含重复记录的Dataframe（整体去重），结果和dropDuplicates不传入指定字段的结果相同。
返回当前不重复的row(行)记录df.distinct()
（2）dropDuplicates:根据指定字段去重
根据指定字段去重。
df.dropDuplicates(Seq("col_name"))

6.聚合

聚合操作调用的是agg方法，该方法有多种调用方式。通常和groupBy方法配合使用。
//输入col，输出dataframe,对id字段求最大，对col2求和。
agg(expers:column*)
df.agg("id" -> "max", "col2" -> "sum")
df.agg(max("age"),avg("salary"))
df.groupBy().agg(max("age"), avg("salary"))

agg(exprs: Map[String, String]) 返回dataframe类型 ，同数学计算求值 map类型的
df.agg(Map("age" -> "max", "salary" -> "avg"))
df.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))
agg(aggExpr: (String, String), aggExprs: (String, String)*) 返回dataframe类型 ，同数学计算求值
df.agg(Map("age" -> "max", "salary" -> "avg"))
df.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))

7. join&union

（1）笛卡尔积
DF1.crossJoin(DF2)
（2）一个字段形式
需要两个DataFrame中有相同的一个列名，默认是”inner”
DF1.join(DF2, "id")
（3）多个字段形式 ，多个字段时最好指定join的类型,有:“full”,”outer”,”full_outer”,”fullouter”代表全连接，“left”,”left_outer”或者”leftouter”代表左连接，“right”,”right_outer”及“rightouter”代表右连接，还有” leftsemi(左半连接)”等等。
val df = df1.join(df2, Seq("key1","key2")) // 基于两个公共字段key1和key的等值连接
DF1.join(DF2, Seq("id", "name")，"inner"） // 多列
val df = df1.join(df2, df1("key1") === df2("key1"), "outer")
DF1.join(DF2, DF1("app_id") === DF2("item_id"), "left")

（4）当两个字段名字不一样时候
DF1.join(DF2 , DF1("id" ) === DF2( "ID")，"inner")

8.1 union

DF1.union(DF2)

8 交叉：

获取两个dataframe共有的纪录
df.intersect(df.limit(1)).show
获取一个dataframe中有另一个daraframe中没有的纪录
df.except(df.limit(1)).show

9.生成一个dataframe&&重命名Dataframe中指定字段名

import sqlContext.implicits._
val df = Seq(
  (12, "First Value", java.sql.Date.valueOf("2010-01-01")),
  (234, "Second Value", java.sql.Date.valueOf("2010-02-01")),
(224, "Second Value", java.sql.Date.valueOf("2010-02-01"))
).toDF("int_column", "string_column", "date_column")

指定字段名不存在则不进行操作，id字段名改为ID
df.withColumnRenamed("id","ID")

10. 读取数据用“|”切分，转成rdd，再把rdd(Array(s0,s1,s2))转化为dataframe

例如： 123123|闪电球|label_2

  case class firstlevel(id: String, name: String, result: String)
## “|”需要两个转义字符
  val textFile_DF = sc.textFile("/user/result")
          .map(v => v.toString.split("\\|"))
          .map{case Array(s0, s1, s2) => firstlevel(s0, s1, s2)}.toDF

11. spark sql和DataFrame中使用UDF（User Defined Function），用户自定义函数

11.1.测试数据

构造测试数据-（name，age）
val userData = Array(("Leo",16),("Marry",21),("Jack",14),("Tom",18))

创建测试df

val user_df = spark.createDataFrame(userData).toDF("name","age")
user_df.show(false)

注册user表
user_df.createOrReplaceTempView("user")

11.2.Spark Sql中的用法

11.2.1 通过匿名函数注册UDF
下面的UDF的功能是计算某列的长度，该列的类型为String
11.2.1.1 注册
spark.udf.register("strLen",(str:String) => str.length())

11.2.2 通过实名函数注册UDF
实名函数注册有点不同，要在后面加 ” _”(空格+下划线)
定义一个实名函数

 def isAdult(age: Int) = {
  if (age < 18) {
    false
  } else {
    true
  }
}

11.2.2.1 注册
spark.udf.register("isAdult", isAdult _)
11.2.2.2 使用
spark.sql("select name,strLen(name) as name_len from user").show

11.3.DataFrame用法
DataFrame的udf方法虽然和Spark Sql的名字一样，但是属于不同的类，它在org.apache.spark.sql.functions里
11.3.1 注册

import org.apache.spark.sql.functions._
//注册自定义函数（通过匿名函数）
val strLen = udf((str: String) => str.length())
//注册自定义函数（通过实名函数）
val udf_isAdult = udf(isAdult _)

11.3.2 使用
可通过withColumn和select使用
给user表添加两列。

//通过withColumn添加列
user_df.withColumn("name_len", strLen(col("name"))).withColumn("isAdult", udf_isAdult(col("age"))).show
//通过select添加列
user_df.select(col("*"), strLen(col("name")) as "name_len", udf_isAdult(col("age")) as "isAdult").show

12.添加一列

添加一列有很多方法：

12.1. withColumn

通常使用withColumn，他会先判断DataFrame里有没有这个列名，如果有的话就会替换掉原来的列，没有的话就用调用select方法增加一列，所以如果我们的需求是增加一列的话，但这个函数只能对原有列做处理生成新列。两者实现的功能一样，且最终都是调用select方法，但是withColumn会提前做一些判断处理，所以withColumn的性能不如select好。
df.withColumn("Column Name", value)
df.withColumn("feature", set_feature(imei_label_feature_df("feature")))

12.2 使用udf

可以用udf写自定义函数新增列

import org.apache.spark.sql.functions.udf
// 新建一个dataFrame
val tempDataFrame = spark.createDataFrame(Seq(
  ("a", "asf"),
  ("b", "2143"),
  ("c", "rfds")
)).toDF("id", "content")
// 自定义udf的函数
val code = (arg: String) => {
      if (arg.getClass.getName == "java.lang.String") 1 else 0
    }

val addCol = udf(code)
// 增加一列
val addColDataframe = tempDataFrame.withColumn("addcol", addCol(tempDataFrame("id")))
addColDataframe.show(10, false)

12.3 添加常量列时，需要使用lit

例如：添加所有行的值均为0的一列
df.withColumn("Column Name", lit(0))
例如：添加所有行的值均为字符串”imei”的一列
df.withColumn("Column Name", lit("imei"))

Spark 2.2引入了typedLit来支持Seq、list等等

import org.apache.spark.sql.functions.typedLit

df.withColumn("some_array", typedLit(Seq(1, 2, 3)))
df.withColumn("some_struct", typedLit(("foo", 1, .0.3)))
df.withColumn("some_map", typedLit(Map("key1" -> 1, "key2" -> 2)))

另外可以使用alias来改变列名，类似sql中的as

df.withColumn(
    "some_struct",
    struct(lit("foo").alias("x"), lit(1).alias("y"), lit(0.3).alias("z"))
 )

可以使用cast函数来改变类型

df.withColumn(
    "some_struct", 
    struct(lit("foo"), lit(1), lit(0.3)).cast("struct<x: string, y: integer, z: double>")
 )

12.3 添加值为null的一列

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
val df = spark.createDataFrame(List(
  (1.2, 1),
  (3.1, 2)))
  .toDF("col1", "col2")
val udf_null = udf((s: Any) => null)
val df_res = df.withColumn("col_name", udf_null(col("col1")).cast(types. StringType))
df_res.show

13.查看schema，更改一列的列名

df.printSchema()
df.withColumnRenamed("x", "y")

14. 过滤操作(filter and where)

14.1.filter & filterNot

filter方法返回了所有使假设条件为真的集合元素组成的新集合。还有一个方法filterNot，可以返回所有使假设条件返回false的元素组成的新集合，用法一样。

val df = sc.parallelize(Seq((“a”, 2), (“b”, 5), (“c”, 2), (“d”, 3), (“e”, 1))).toDF(“id”, “num”)
整数类型
逻辑运算符：>, <, ===

val number = 2;
df.filter(df.col("num") === 2)
df.filter(df("num") === 2)
df.filter($"num">number)    //传递参数过滤(新版本scala可能不能用)
df.filter($"num"<2)

或者

df.filter("num=2")
df.filter("num>2")
df.filter("num is null")

字符串类型
df.filter($"id".equalTo("a"))

传递参数过滤

val str = s"a"

df.filter($"id"equalTo(str))

当dataframe没有字段名时，可以用默认的字段名[_1, _2, …..]来进行判断

多条件判断
逻辑连接符 &&（并）、||（或）

df.filter($"num"===2 && $"id".equalTo("a")
df.filter($"num"===1 || $"num"===3)

14.2.where

df.where("id = 1 or sex = 'male' ").show
用法和filter相同

15.转化列类型

单列转换

import org.apache.spark.sql.types._
//import org.apache.spark.sql.types.{DoubleType, IntegerType}

val data = Array(("1", "2", "3", "4", "5"), ("6", "7", "8", "9", "10"))
val df = spark.createDataFrame(data).toDF("col1", "col2", "col3", "col4", "col5")

import org.apache.spark.sql.functions._
df.select(col("col1").cast(DoubleType)).show()

使用withColumn循环多列转换(遍历)

import org.apache.spark.sql.types
var df1 = df
val colNames = df.columns

for (colName <- colNames) {
  df1 = df1.withColumn(colName, col(colName).cast(types.DoubleType))
// 有些type名称会和其他类里的名称起冲突，例如StringType，调用的时候手动加上类名如types.StringType调用。
}
df1.show()

16.创建一个空dataframe

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types
import org.apache.spark.sql.Row
import org.apache.spark.rdd.EmptyRDD

/**
 * Spark创建空DataFrame示例
 */
object EmptyDataFrame {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("EmptyDataFrame").master("local").getOrCreate()

    /**
     * 创建一个空的DataFrame，代表用户
     * 有四列，分别代表ID、名字、年龄、生日
     */
    val colNames = Array("id", "name", "age", "birth")
    //为了简单起见，字段类型都为String
    val schema = StructType(colNames.map(fieldName => StructField(fieldName, types.StringType, true)))
    //主要是利用了spark.sparkContext.emptyRDD
    val emptyDf = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema)

    emptyDf.show

    /**
   也可以给每列指定相对应的类型(使用这种会比较多)
   通过StructType直接指定每个字段的schema
     */
    val schema1 = StructType(
      Seq(
        StructField("id", types.IntegerType, true),
        StructField("name", types.StringType, true),
        StructField("age", types.IntegerType, true),
        StructField("birth", types.StringType, true)))
    val emptyDf1 = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema1)
    emptyDf1.show

    //还有一种空的DataFrame，没有任何行任何列
    spark.emptyDataFrame.show

    spark.stop()
  }
}

17.指定列缺失值填充

    val df = source_df.na.fill(value = 0.0,cols=Array("col1","col2")
    val df1 = source_df.na.fill(value = "male",cols=Array("sex"))

18.行转列

根据c3字段中空格将字段内容分割，分割内容存在c3_字段中
df.explode( "c3" , "c3_" ){time: String => time.split( " " )}

19.对dataframe中的值进行替换

我们需要把第二列中的“Tesla”改成“S”,“Ford”改成“F”

//初始化一个dataframe
//import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.{functions => F}
import org.apache.spark.sql.{DataFrame,SparkSession}

val sparkSession =  SparkSession.builder().enableHiveSupport().getOrCreate()
val rdd = sc.parallelize(
      List( (2012,"Tesla","S"), (1997,"Ford","E350"), (2015,"Chevy","Volt"))
  )
 import sparkSession.implicits._
  val df = rdd.toDF("year","trademark","model_number")

使用functions.udf

val makeSIfTesla = udf {(trademark: String) => 
  if(trademark == "Tesla") "S"  else if(trademark == "Ford")"F"
 else trademark
}
var df1 = df.withColumn("trademark", makeSIfTesla(df("trademark")))

使用functions.when

val df2 = df.withColumn("trademark", F.when(col("trademark").equalTo("Tesla"), "S").when(col("trademark").equalTo("Ford"), "F").otherwise(col("trademark")))

val df3 = df.withColumn("trademark", F.when(col("trademark") === "Tesla", "S").when(col("trademark") === "Ford", "F").otherwise(col("trademark")))

19.注册成表，并进行SparkSQL操作，以及df的map操作

df.show
这个数据代表用户有没有车，label = 1为有车。

• 将DataFrame注册成表
  df.registerTempTable(“people_car_table”)

• 利用sql方法进行SparkSQL操作，选出有车的用户
  val people_car = sparkSession.sql("SELECT imei,label FROM people_car_table WHERE label = 1")

• 将返回结果看作是数据库操作的一行，（0）表示第一列，依次类推
  这里把imei前面加上字符串为“IMEI:”的前缀
  people_car.map(t => "IMEI: " + t(0)).collect().foreach(println)

  
  people_car.map(t => "IMEI: " + t(0)).show
  
  同时改变两列
  people_car.map(t => ("imei:"+t(0),"label:"+t(1))).show
  

• 通过域的名称获取信息,结果和people_car.map(t => “IMEI: ” + t(0)).show一样
  people_car.map(t => "Name: " + t.getAs[String]("imei")).show

上面两张图可以看出dataframe的map操作和rdd基本是一样的，但是dataframe相比rdd的优势在于带有schema，更加直观，使用map以后schema就变化了，需要重新定义schema。

• 通过sql语句读取hive表，dataframe格式
  val car_df = sparkSession.sql(select_sql)

20.数据的拆分与合并

20.1.数据的合并

concat合并列，列类型需要为字符串 
concat_ws按照特殊符号分隔来合并列，返回col格式
连接字符串：
concat_ws("_", field1, field2)，输出结果将会是：“field1_field2”。

数组元素连接：
concat_ws("_", [a,b,c])，输出结果将会是："a_b_c"。
多个列连接：
concat_ws(",",col(col_name1),col(col_name2),col(col_name3))

import org.apache.spark.sql.types._
import org.apache.spark.sql.types
import org.apache.spark.sql.{DataFrame, Row}
//import org.apache.spark.sql.types.{DoubleType, StringType}

val data = Array(("1", "2", "3", "4", "5"), ("6", "7", "8", "9", "10"))
val df = spark.createDataFrame(data).toDF("col1", "col2", "col3", "col4", "col5")

//合并多个列，加在原dataframe的后面
val new_df = df.withColumn("new_col_name", concat_ws(",", col("col1"),col("col2"),col("col3")).cast(types.StringType))

//使用自定义UDF函数
    // 编写udf函数，对行做处理
val separator = "，"
    def mergeCols(merge_row: Row): String = {
        merge_row.toSeq.foldLeft("")(_ + separator + _).substring(1)
    }
    val mergeColsUDF = udf(mergeCols _)
    df.select($"col4",$"col5",mergeColsUDF(struct($"col1", $"col2", $"col3")).as("value")).show

结果：
+----+----+-----+
|col4|col5|value|
+----+----+-----+
|   4|   5|1，2，3|
|   9|  10|6，7，8|
+----+----+-----+

20.2.数据的拆分

把一列（字符串），按照特殊的分隔符号切分成多列

生成数据
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row}
//import org.apache.spark.sql.types.{DoubleType, StringType}

val data = Array(("1", "2", "3", "4", "5"), ("6", "7", "8", "9", "10"))
var df = spark.createDataFrame(data).toDF("col1", "col2", "col3", "col4", "col5")
val separator = "，"
    def mergeCols(merge_row: Row): String = {
        merge_row.toSeq.foldLeft("")(_ + separator + _).substring(1)
    }

    val mergeColsUDF = udf(mergeCols _)
   df = df.select(mergeColsUDF(struct($"col1", $"col2", $"col3",$"col4",$"col5")).as("value"))

使用内置函数split，然后遍历添加列
    val separator = "，"
    lazy val first = df.first()
使用lazy val first，原因是构造顺序问题，在使用.length等方法时，first可能还没有被赋值，所以先使用lazy val生成，等到用到这个变量的时候再初始化。
    val numAttrs = first.toString().split(separator).length
tabulate返回指定长度数组，每个数组元素为指定函数的返回值，默认从0开始。
    val attrs = Array.tabulate(numAttrs)(n => "col_" + n)
    //按指定分隔符拆分value列，生成splitCols列
    var newDF = df.withColumn("splitCols", split($"value", separator))
    attrs.zipWithIndex.foreach(x => {
      newDF = newDF.withColumn(x._1, $"splitCols".getItem(x._2))
    })
    newDF.show()

结果：
+----------+----------------+-----+-----+-----+-----+-----+
|     value|       splitCols|col_0|col_1|col_2|col_3|col_4|
+----------+----------------+-----+-----+-----+-----+-----+
| 1，2，3，4，5| [1, 2, 3, 4, 5]|    1|    2|    3|    4|    5|
|6，7，8，9，10|[6, 7, 8, 9, 10]|    6|    7|    8|    9|   10|
+----------+----------------+-----+-----+-----+-----+-----+

21.查看分区

df.rdd.getNumPartitions

22.Spark SQL符合类型struct

df.selectExpr("(col1,col2) as new_col","*").show(1)
df.selectExpr("struct(col1,col2) as new_col","*").show(1)
val complexDF=df.select(struct("col1","col2").as("new_col3"))
complexDF.createOrReplaceTempView("complexDF")
complexDF.select("new_col3.col1").show(1)
complexDF.select(col("new_col3").getField("col2")).show(1)
complexDF.select("new_col3.*").show(1)
spark.sql("select new_col3.* from complexDF").show(1)

Reference

https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html
https://stackoverflow.com/questions/32357774/scala-how-can-i-replace-value-in-dataframes-using-scala
https://stackoverflow.com/questions/32788322/how-to-add-a-constant-column-in-a-spark-dataframe
https://www.cnblogs.com/xiaoma0529/p/7126923.html
https://dongkelun.com/2018/08/14/sparkEmptyDataFrame/
https://dongkelun.com/2018/08/02/sparkUDF/
https://blog.csdn.net/sinat_36121406/article/details/82755516