一、dataframe操作大全

增/删/改/查/合并/统计与数据处理:

spark左外连接：

structField、structType、schame：

dataframe、dataset、sql.dataframe：

 创建dataframe：

二、dataframe的filter用法

val df = sc.parallelize(Seq(("a", 1), ("a", 2), ("b", 2), ("b", 3), ("c", 1))).toDF("id", "num")

1、对整数类型过滤

（1）逻辑运算符：>, <, ===

df.filter($"num"===2)df.filter($"num">2)df.filter($"num"<2)

或者

df.filter("num=2")df.filter("num>2")df.filter("num<2")

（2）传递参数过滤

val ind:Int=2;df.filter($"num"===ind)df.filter($"num">ind)df.filter($"num"

2、对字符串过滤

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

（1）传递参数过滤

val str = s"a"df.filter($"id"equalTo(str))

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

3、多条件判断

逻辑连接符 &&（并）、||（或）

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

 

 

三、DataFrame和DataSet[T]无法使用map的问题：Unable to find encoder for type stored in a Dataset

spark2.0以后的版本采用的是新的分布式数据集DataSet，其中DataFrame是DataSet[Row]的别名形式。

 

DataSet数据集在使用sql()时，无法使用map，flatMap等转换算子的解决办法：

方法一：要想对dataset进行操作，需要进行相应的encode操作。要进行map操作，要先定义一个Encoder。特别是官网给的例子：

// No pre-defined encoders for Dataset[Map[K,V]], define explicitly implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[String, Any]] // Primitive types and case classes can be also defined as // implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder() // row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T] teenagersDF.map(teenager => teenager.getValuesMap[Any](List("name", "age"))).collect() // Array(Map("name" -> "Justin", "age" -> 19))

方法二：（不推荐这种方法）为了更简单一些，dataset也提供了转化RDD的操作。因此只需要将之前dataframe.map 在中间修改为：dataframe.rdd.map即可。