Spark學習之Dataset (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 進行操做

val ds = spark.range(10)
    ds.transform(item => item.withColumn("double",'id *2))
      .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

因此, 使用 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 用來選擇某些列出如今結果集中

（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()

  }

（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",'name).show()
    ds.withColumn("name_joke",'name==="").show()
  }

（4）withColumnRenamed

方法描述：修改列名

@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.withColumnRenamed("name","new_name").show()
  }

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()
  }