有型別操作
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()
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/251613.html
標籤：大數據