Spark学习之Dataset (DataFrame) 的基础操作

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

原文转载：http://www.shaoqun.com/a/512303.html

跨境电商：https://www.ikjzd.com/

review：https://www.ikjzd.com/w/2735

一淘网比价平台：https://www.ikjzd.com/w/1698

Dataset(DataFrame)的常见有类型操作、无类型操作有类型操作1.转换类型的操作转换类型的操作主要包含：flatMap、map、mapPartitions、transform、as（1）flatMap方法描述：通过flatMap可以将一条数据转为一个数组,后再展开这个数组放入Datasetvalds:Dataset[String]=Seq("hellospark",
兰亭集势：兰亭集势
走秀网：走秀网
广州番禺莲花水乡门票多少钱？：广州番禺莲花水乡门票多少钱？
良心干货：做亚马逊必备的8款小工具！：良心干货：做亚马逊必备的8款小工具！
北海海上观光团购价？北海环岛游价格？：北海海上观光团购价？北海环岛游价格？