Spark Dataset (DataFrame) 常见操作

小鱼儿 2023-01-10 14:44 200阅读 0赞

参考文章：[Spark学习之Dataset (DataFrame) 的基础操作][Spark_Dataset _DataFrame_]

[Spark创建DataFrame的三种方法][Spark_DataFrame]

## 一. 有类型操作 ##

### 1.转换类型的操作 ###

转换类型的操作主要包含：***flatMap、map、mapPartitions、transform、as***

**（1）flatMap**

方法描述：通过 `flatMap `可以将一条数据转为一个数组, 后再展开这个数组放入 Dataset

val ds: Dataset[String] = Seq("hello spark","hello hadoop").toDS()
        ds.flatMap(item => item.split(" ")).show()

**（2）map**

方法描述：`map` 可以将数据集中每条数据转为另一种形式（通过传入计算函数来实现）

val ds2 = Seq(Person("zhangsan",15),Person("lisi",20)).toDS()
        ds2.map(item => (Person)(item.name,item.age*2)).show()

**（3）mapPartitions**

方法描述：`mapPartitions` 和 map 一样, 但是 map 的处理单位是每条数据, `mapPartitions` 的处理单位是每个分区

ds2.mapPartitions(
          //iter 不能大到每个Executor的内存放不下，不然就会OOM
          //对每个元素进行转换，后生成一个新的集合
          iter => {
            val result = iter.map(item => Person(item.name,item.age*2))
            result
          }
        ).show()

**（4）transform**

方法描述：`map `和 `mapPartitions` 以及`transform`都是转换, `map` 和` mapPartitions` 是针对数据, 而 `transform `是针对整个数据集, 这种方式最大的区别就是 `transform` 可以直接拿到` Dataset` 进行操作

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE4ODAwNDYz_size_16_color_FFFFFF_t_70][]

val ds = spark.range(10)
        ds.transform(item => item.withColumn("double", column("age") + 3))
          .show()

**（5）as**

方法描述：`as[Type] `算子的主要作用是将弱类型的` Dataset` 转为强类型的 `Dataset`, 它有很多适用场景, 但是最常见的还是在读取数据的时候, 因为`DataFrameReader`体系大部分情况下是将读出来的数据转换为 `DataFrame`的形式, 如果后续需要使用 `Dataset` 的强类型 `API`, 则需要将 `DataFrame` 转为 `Dataset`. 可以使用 `as[Type]` 算子完成这种操作

@Test
      def as(): Unit = {
        var schema = StructType(
          Seq(
            StructField("name",StringType),
            StructField("age",IntegerType),
            StructField("gpa",FloatType)
          )
        )
        val df: DataFrame = spark.read
          .schema(schema)
          .option("delimiter","\t")
          .csv("dataset/studenttab10k")
    
        val ds: Dataset[Student] = df.as[Student]
    
        ds.show()
      }

### 2 .过滤类型的操作 ###

过滤类型的操作主要包含：**filter**

**（1）filter**

方法描述：用来按照条件过滤数据集

@Test
      def filter(): Unit = {
        import spark.implicits._
        val ds = Seq(Person("zhangsan",15),Person("lisi",20)).toDS()
        ds.filter(item => item.age>15)
          .show()
      }

### 3.集合类型的操作 ###

集合类型的操作主要包含：**groupBykey**

**（1）groupByKey**

方法描述：`grouByKey` 算子的返回结果是 `KeyValueGroupedDataset`, 而不是一个 `Dataset`, 所以必须要先经过 `KeyValueGroupedDataset` 中的方法进行聚合, 再转回 `Dataset`, 才能使用 `Action` 得出结果。  
其实这也印证了分组后必须聚合的道理

@Test
      def groupByKey(): Unit= {
        import spark.implicits._
        val ds = Seq(Person("zhangsan",15),Person("lisi",20)).toDS()
    
        val grouped: KeyValueGroupedDataset[String, Person] = ds.groupByKey(item => item.name)
        val result: Dataset[(String, Long)] = grouped.count()
        result.show()
      }

### 4.切分类型的操作 ###

切分类型的操作主要包含：**randomSplit、sample**

**（1）randomSplit**

方法描述：`randomSplit` 会按照传入的权重随机将一个 `Dataset` 分为多个 `Dataset`, 传入 `randomSplit` 的数组有多少个权重, 最终就会生成多少个 `Dataset`, 这些权重的加倍和应该为 1, 否则将被标准化

@Test
    def split(): Unit = {
      val ds = spark.range(15)
      val datasets: Array[Dataset[lang.Long]] = ds.randomSplit(Array(5,2,3))
      datasets.foreach(_.show())
    }

**（2）sample**

方法描述：`sample` 会随机在 `Dataset` 中抽样

@Tests
    def split(): Unit = {
      val ds = spark.range(15)
      ds.sample(withReplacement = false,fraction = 0.4).show()
    }

### 5.排序类型的操作 ###

排序类型的操作主要包含：***orderBy、sort***

**（1）orderBy**

方法描述：`orderBy` 配合 `Column` 的 `API`, 可以实现正反序排列

@Test
    def sort(): Unit = {
      val ds = Seq(Person("zhangsan",15),Person("lisi",20),Person("ss",5)).toDS()
      ds.orderBy('age.desc_nulls_first).show()
    }

**（2）sort**

方法描述：其实 `orderBy` 是 `sort` 的别名, 所以它们所实现的功能是一样的

@Test
    def sort(): Unit = {
      val ds = Seq(Person("zhangsan",15),Person("lisi",20),Person("ss",5)).toDS()
    
      ds.sort('age.asc).show()
    }

### 6.分区类型的操作 ###

分区类型的操作主要包含：***coalesce、repartitions***

**（1）coalesce**

方法描述：减少分区, 此算子和 `RDD` 中的 `coalesce` 不同, `Dataset` 中的 `coalesce` 只能减少分区数, `coalesce` 会直接创建一个逻辑操作, 并且设置 `Shuffle` 为 **false**

val ds = spark.range(15)
    ds.coalesce(1).explain(true)

**（2）repartitions**

方法描述：`repartitions` 有两个作用, 一个是重分区到特定的分区数, 另一个是按照某一列来分区, 类似于 `SQL` 中的 `DISTRIBUTE BY`

val ds = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS()
    ds.repartition(4)
    ds.repartition('name)

### 7.去重类型的操作 ###

去重类型的操作主要包含：***dropDuplicates、distinct\****

**（1）dropDuplicates**

方法描述：使用 `dropDuplicates` 可以去掉某一些列中重复的行

val ds =  Seq(Person("张三",15),Person("张三",15),Person("李四",20)).toDS()
    	ds.dropDuplicates("age").show()

**（2）distinct**

方法描述：当 `dropDuplicates` 中没有传入列名的时候, 其含义是根据所有列去重, `dropDuplicates() `方法还有一个别名, 叫做 distinct

![2019093015513369.png][]

所以, 使用 distinct 也可以去重, 并且只能根据所有的列来去重

val ds =  Seq(Person("张三",15),Person("张三",15),Person("李四",20)).toDS()
        ds.distinct().show()

### 8.集合类型的操作 ###

集合类型的操作主要包含：***\*except、intersect、union、limit\****

**（1）except**

方法描述：`except` 和 `SQL` 语句中的 `except` 一个意思, 是求得 `ds1` 中不存在于 `ds2` 中的数据, 其实就是**差集**

@Test
    def collection(): Unit = {
      val ds1 = spark.range(10)
      val ds2 = spark.range(5,15)
    
      ds1.except(ds2).show()
    }

**（2）intersect**

方法描述：求得两个集合的交集

@Test
    def collection(): Unit = {
      val ds1 = spark.range(10)
      val ds2 = spark.range(5,15)
    
    
      ds1.intersect(ds2).show()
    }

**（3）union**

方法描述：求得两个集合的并集

@Test
    def collection(): Unit = {
      val ds1 = spark.range(10)
      val ds2 = spark.range(5,15)
    
      ds1.union(ds2).show()
    }

**（4）limit**

方法描述：限制结果集数量

@Test
    def collection(): Unit = {
      val ds1 = spark.range(10)
      val ds2 = spark.range(5,15)
    
      ds1.limit(5).show()
    }

## 二. 无类型操作 ##

### 1. 选择类型的操作 ###

选择类型的操作主要包含：***select、selectExpr、withColumn、withColumnRenamed***

**（1）select**

方法描述：`select` 用来选择某些列出现在结果集中

df.select("count", "new_column_name")

**（2）selectExpr**

方法描述：在 `SQL` 语句中, 经常可以在 `select` 子句中使用 count(age), rand() 等函数, 在 `selectExpr` 中就可以使用这样的 `SQL` 表达式, 同时使用 `select` 配合 `expr` 函数也可以做到类似的效果

val spark = SparkSession.builder()
        .master("local[6]")
        .appName(this.getClass.getName)
        .getOrCreate()
      import spark.implicits._
    
      @Test
      def select(): Unit = {
        val ds = Seq(Person("zhangsan",15),Person("lisi",20),Person("cq",18)).toDS()
    
        ds.selectExpr("sum(age)").show()
        ds.select('name).show()
    
        import org.apache.spark.sql.functions._
        ds.select(expr("sum(age)")).show()
    
        // other exam: recs is a dataframe
        val recs = ratings.as[(Int, Int, Float)]
                      .groupByKey(_._1)
                      .agg(topKAggregator.toColumn)
                      .toDF("id", "recommendations")
        recs.select($"id".as(srcOutputColumn), $"recommendations".cast(arrayType))
    
        ds.selectExpr("appid as newappid").show()
        df.selectExpr("count(distinct(appid)) as count1", "count(distinct(brand)) as count2").show()
      }

**（3）withColumn**

方法描述：通过 Column 对象在 Dataset 中创建一个新的列或者修改原来的列

@Test
    def column(): Unit = {
      val ds = Seq(Person("zhangsan",15),Person("lisi",20),Person("cq",18)).toDS()
      
      // select rand() from ..
      // 如果想使用函数的功能，使用 function.xx   使用表达式 expr("...")
      import org.apache.spark.sql.functions._
      ds.withColumn("random", expr("rand()")).show()
      ds.withColumn("name_new", col("name")).show()
      ds.withColumn("name_new2", lit("name")).show()
      ds.withColumn("name_joke", column("age") + 3).show()
    
      ds..withColumn("upsert_time", lit(System.currentTimeMillis()))
    }

**（4）withColumnRenamed**

方法描述：修改列名

###  2. 剪切类型的操作 ###

剪切类型的操作主要包含：***\*drop\****

**（1）drop**

方法描述：剪掉某个列

@Test
    def groupBy(): Unit = {
      val ds = Seq(Person("zhangsan",15),Person("lisi",20),Person("lisi",18)).toDS()
    
      import org.apache.spark.sql.functions._
      ds.drop("name").show()
    }

### 3.聚合类型的操作 ###

聚合类型的操作主要包含：***groupBy***

**（1）groupBy**

方法描述：按照给定的行进行分组

@Test
    def groupBy(): Unit = {
      val ds = Seq(Person("zhangsan",15),Person("lisi",20),Person("lisi",18)).toDS()
    
      import org.apache.spark.sql.functions._
      ds.groupBy('name).agg(mean("age")).show()
    }
    
    // 示例1:
    data.groupBy("gender").agg(count($"age"),max($"age").as("maxAge"), avg($"age").as("avgAge")).show
    +------+----------+------+------+                                              
    |gender|count(age)|maxAge|avgAge|
    +------+----------+------+------+
    |female|         5|  32.0|  29.0|
    |  male|         5|  57.0|  39.0|
    +------+----------+------+------+
     
    // 示例2: 
    data.groupBy("gender").agg("age"->"count","age" -> "max", "age" -> "avg").show
    +------+----------+--------+--------+                                          
    |gender|count(age)|max(age)|avg(age)|
    +------+----------+--------+--------+
    |female|         5|    32.0|    29.0|
    |  male|         5|    57.0|    39.0|
    +------+----------+--------+--------+

[Spark_Dataset _DataFrame_]: https://www.cnblogs.com/xp-thebest/p/14300479.html
[Spark_DataFrame]: https://blog.csdn.net/martin_liang/article/details/79748503
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE4ODAwNDYz_size_16_color_FFFFFF_t_70]: /images/20221119/d9019ef5950f40dbb519c07da38b1d1c.png
[2019093015513369.png]: https://img-blog.csdnimg.cn/2019093015513369.png