转换
|
flatMap
|
通过 flatMap 可以将一条数据转为一个数组, 后再展开这个数组放入 Dataset
import spark.implicits._
val ds = Seq("hello world", "hello pc").toDS()
ds.flatMap( _.split(" ") ).show()
|
map
|
import spark.implicits._
val ds = Seq(Person("zhangsan", 15), Person("lisi", 15)).toDS()
ds.map( person => Person(person.name, person.age * 2) ).show()
|
mapPartitions
|
mapPartitions 和 map 一样, 但是 map 的处理单位是每条数据, mapPartitions 的处理单位是每个分区
import spark.implicits._
val ds = Seq(Person("zhangsan", 15), Person("lisi", 15)).toDS()
ds.mapPartitions( iter =>
val returnValue = iter.map(
item => Person(item.name, item.age * 2)
)
returnValue
)
.show()
|
transform
|
map 和 mapPartitions 以及 transform 都是转换, map 和 mapPartitions 是针对数据, 而 transform 是针对整个数据集, 这种方式最大的区别就是 transform 可以直接拿到 Dataset 进行操作
import spark.implicits._
val ds = spark.range(5)
ds.transform( dataset => dataset.withColumn("doubled", ‘id * 2) )
|
as
|
as[Type] 算子的主要作用是将弱类型的 Dataset 转为强类型的 Dataset , 它有很多适用场景, 但是最常见的还是在读取数据的时候, 因为 DataFrameReader 体系大部分情况下是将读出来的数据转换为 DataFrame 的形式, 如果后续需要使用 Dataset 的强类型 API , 则需要将 DataFrame 转为 Dataset . 可以使用 as[Type] 算子完成这种操作
import spark.implicits._
val structType = StructType(
Seq(
StructField("name", StringType),
StructField("age", IntegerType),
StructField("gpa", FloatType)
)
)
val sourceDF = spark.read
.schema(structType)
.option("delimiter", "\t")
.csv("dataset/studenttab10k")
val dataset = sourceDF.as[Student]
dataset.show()
|
过滤
|
filter
|
import spark.implicits._
val ds = Seq(Person("zhangsan", 15), Person("lisi", 15)).toDS()
ds.filter( person => person.name == "lisi" ).show()
|
聚合
|
groupByKey
|
grouByKey 算子的返回结果是 KeyValueGroupedDataset , 而不是一个 Dataset , 所以必须要先经过 KeyValueGroupedDataset 中的方法进行聚合, 再转回 Dataset , 才能使用 Action 得出结果
import spark.implicits._
val ds = Seq(Person("zhangsan", 15), Person("zhangsan", 15), Person("lisi", 15)).toDS()
ds.groupByKey( person => person.name ).count().show()
|
切分
|
randomSplit
|
randomSplit 会按照传入的权重随机将一个 Dataset 分为多个 Dataset , 传入 randomSplit 的数组有多少个权重, 最终就会生成多少个 Dataset , 这些权重的加倍和应该为 1, 否则将被标准化
val ds = spark.range(15)
val datasets: Array[Dataset[lang.Long]] = ds.randomSplit(Array[Double](2, 3))
datasets.foreach(dataset => dataset.show())
|
sample
|
val ds = spark.range(15)
ds.sample(withReplacement = false, fraction = 0.4).show()
|
排序
|
orderBy
|
orderBy 配合 Column 的 API , 可以实现正反序排列
import spark.implicits._
val ds = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS()
ds.orderBy("age").show()
ds.orderBy(‘age.desc).show()
|
sort
|
其实 orderBy 是 sort 的别名, 所以它们所实现的功能是一样的
import spark.implicits._
val ds = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS()
ds.sort(‘age.desc).show()
|
分区
|
coalesce
|
减少分区, 此算子和 RDD 中的 coalesce 不同, Dataset 中的 coalesce 只能减少分区数, coalesce 会直接创建一个逻辑操作, 并且设置 Shuffle 为 false
val ds = spark.range(15)
ds.coalesce(1).explain(true)
|
repartitions
|
repartitions 有两个作用, 一个是重分区到特定的分区数, 另一个是按照某一列来分区, 类似于 SQL 中的 DISTRIBUTE BY
val ds = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS()
ds.repartition(4)
ds.repartition(‘name)
|
去重
|
dropDuplicates
|
使用 dropDuplicates 可以去掉某一些列中重复的行
import spark.implicits._
val ds = spark.createDataset(Seq(Person("zhangsan", 15), Person("zhangsan", 15), Person("lisi", 15)))
ds.dropDuplicates("age").show()
|
distinct
|
当 dropDuplicates 中没有传入列名的时候, 其含义是根据所有列去重, dropDuplicates() 方法还有一个别名, 叫做 distinct
所以, 使用 distinct 也可以去重, 并且只能根据所有的列来去重
import spark.implicits._
val ds = spark.createDataset(Seq(Person("zhangsan", 15), Person("zhangsan", 15), Person("lisi", 15)))
ds.distinct().show()
|
集合操作
|
except
|
except 和 SQL 语句中的 except 一个意思, 是求得 ds1 中不存在于 ds2 中的数据, 其实就是差集
val ds1 = spark.range(1, 10)
val ds2 = spark.range(5, 15)
ds1.except(ds2).show()
|
intersect
|
val ds1 = spark.range(1, 10)
val ds2 = spark.range(5, 15)
ds1.intersect(ds2).show()
|
union
|
val ds1 = spark.range(1, 10)
val ds2 = spark.range(5, 15)
ds1.union(ds2).show()
|
limit
|
val ds = spark.range(1, 10)
ds.limit(3).show()
|