0


大数据-95 Spark 集群 SparkSQL Action与Transformation操作 详细解释与测试案例

点一下关注吧!!!非常感谢!!持续更新!!!

目前已经更新到了:

  • Hadoop(已更完)
  • HDFS(已更完)
  • MapReduce(已更完)
  • Hive(已更完)
  • Flume(已更完)
  • Sqoop(已更完)
  • Zookeeper(已更完)
  • HBase(已更完)
  • Redis (已更完)
  • Kafka(已更完)
  • Spark(正在更新!)

章节内容

上节完成的内容如下:

  • SparkSession
  • RDD、DataFrame、DataSet
  • 三者之间互相转换 详细解释在这里插入图片描述

核心操作

Transformation(转换操作)

定义:
Transformation是懒执行的操作,意味着这些操作在调用时并不会立即执行计算,而是会生成一个新的数据集(或RDD),它们描述了从输入数据到输出数据的转换逻辑。Transformation的计算会被延迟,直到遇到一个Action操作时才会真正触发执行。

常见操作:

  • select(): 从DataFrame中选择列。
  • filter(): 过滤掉不符合条件的行。
  • join(): 连接两个DataFrame。
  • groupBy(): 对数据进行分组。
  • agg(): 聚合操作。

Action(行动操作)

定义:
Action操作会触发Spark的计算并返回结果。与Transformation不同,Action操作会执行整个计算逻辑,并产生最终的输出,如将结果写入外部存储或将数据返回给驱动程序。

常见操作:

  • show(): 显示DataFrame的内容。
  • collect(): 将DataFrame的数据收集到驱动程序上,作为本地集合返回。
  • count(): 计算DataFrame中的行数。
  • write(): 将DataFrame的数据写入外部存储(如HDFS、S3、数据库等)。
  • take(): 返回DataFrame的前n行数据。

Action操作

与RDD类似的操作

  • show
  • collect
  • collectAsList
  • head
  • first
  • count
  • take
  • takeAsList
  • reduce

与结构相关

  • printSchema
  • explain
  • columns
  • dtypes
  • col

生成数据

保存并上传到服务器上

EMPNO,ENAME,JOB,MGR,HIREDATE,SAL,COMM,DEPTNO
7369,SMITH,CLERK,7902,2001-01-02 22:12:13,800,,20
7499,ALLEN,SALESMAN,7698,2002-01-02 22:12:13,1600,300,30
7521,WARD,SALESMAN,7698,2003-01-02 22:12:13,1250,500,30
7566,JONES,MANAGER,7839,2004-01-02 22:12:13,2975,,20
7654,MARTIN,SALESMAN,7698,2005-01-02 22:12:13,1250,1400,30
7698,BLAKE,MANAGER,7839,2005-04-02 22:12:13,2850,,30
7782,CLARK,MANAGER,7839,2006-03-02 22:12:13,2450,,10
7788,SCOTT,ANALYST,7566,2007-03-02 22:12:13,3000,,20
7839,KING,PRESIDENT,,2006-03-02 22:12:13,5000,,10
7844,TURNER,SALESMAN,7698,2009-07-02 22:12:13,1500,0,30
7876,ADAMS,CLERK,7788,2010-05-02 22:12:13,1100,,20
7900,JAMES,CLERK,7698,2011-06-02 22:12:13,950,,30
7902,FORD,ANALYST,7566,2011-07-02 22:12:13,3000,,20
7934,MILLER,CLERK,7782,2012-11-02 22:12:13,1300,,10

写入内容如下图所示:
在这里插入图片描述

测试运行

我们进入 spark-shell 进行测试

// 处理头,使用自动类型推断val df1 = spark.read.option("header",true).option("infershema","true").csv("test_spark_03.txt")

df1.count
// 缺省显示20行
df1.union(df1).show()// 显示2行
df1.show(2)

执行结果如下图所示:
在这里插入图片描述
继续进行测试:

// 不截断字符
df1.toJSON.show(false)// 显示10行 不截断字符
df1.toJSON.show(10,false)

运行结果如下图所示:
在这里插入图片描述
继续进行测试:

// collect 返回数组 Array[Row]val c1 = df1.collect()// collectAsList 返回List Lits[Row]val c2 = df1.collectAsList()// 返回 Rowval h1 = df1.head()val f1 = df1.first()// 返回 Array[Row]val h2 = df1.head(3)val f2 = df1.take(3)// 返回 List[Row]val t2 = df1.takeAsList(2)

运行结果如下图所示:
在这里插入图片描述
继续进行测试:

// 结构属性// 查看列名
df1.columns
// 查看列名和类型
df1.dtypes
// 查看执行计划
df1.explain()// 获取某个列
df1.col("ENAME")// 常用
df1.printSchema

运行结果如下图所示:
在这里插入图片描述

Transformation 操作

  • RDD 类似的操作
  • 持久化/缓存 与 checkpoint
  • select
  • where
  • group by / 聚合
  • order by
  • join
  • 集合操作
  • 空值操作(函数)
  • 函数

与RDD类似的操作

  • map
  • filter
  • flatMap
  • mapPartitions
  • sample
  • randomSplit
  • limt
  • distinct
  • dropDuplicates
  • describe

我们进行测试:

val df1 = spark.read.csv("/opt/wzk/data/people1.csv")// 获取第1列
df1.map(row => row.getAs[String](0)).show

// randomSplit 将DF、DS按给定参数分成多份val df2 = df1.randomSplit(Array(0.5,0.6,0.7))
df2(0).count
df2(1).count
df2(2).count

测试结果如下图:
在这里插入图片描述

我们继续进行测试:

// 取10行数据生成新的Datasetval df2 = df1.limit(10)// distinct 去重val df2 = df1.union(df1)
df2.distinct.count

// dropDuplicates 按列值去重
df2.dropDuplicates.show
df2.dropDuplicates("_c0").show

执行结果如下图:
在这里插入图片描述

存储相关

  • cacheTable
  • persist
  • checkpoint
  • unpersist
  • cache

备注:Dataset默认的存储级别是 MEMEORY_AND_DISK

spark.sparkContext.setCheckpointDir("hdfs://h121.wzk.icu:9000/checkpoint")

df1.show()
df1.checkpoint()
df1.cache()importorg.apache.spark.storage.StorageLevel
df1.persist(StorageLevel.MEMORY_ONLY)
df1.count()
df1.unpersist(true)

执行结果如下图所示:
在这里插入图片描述

select相关

  • 列的多种表示
  • select
  • selectExpr

启动 Spark-Shell 继续进行测试

// 这里注意 option("header", "true") 自动解析一下表头val df1 = spark.read.option("header","true").csv("/opt/wzk/data/people1.csv")// $ col() 等等 不可以混用!!!(有解决方法,但是建议不混用!!!)// 可以多种形式获取到列
df1.select($"name", $"age", $"job").show

执行结果如下图所示:
在这里插入图片描述

继续进行测试

df1.select("name","age","job").show(3)
df1.select(col("name"), col("age"), col("job")).show(3)
df1.select($"name", $"age"+1000, $"job").show(5)

运行结果如下图所示:
在这里插入图片描述

where相关

接着对上述内容进行测试:

df1.filter("age > 25").show
df1.filter("age > 25 and name == 'wzk18'").show
df1.where("age > 25").show
df1.where("age > 25 and name == 'wzk19'").show

运行测试结果如下图:
在这里插入图片描述

groupBy相关

  • groupBy
  • agg
  • max
  • min
  • avg
  • sum
  • count

进行测试:

// 由于我的字段中没有数值类型的,就不做测试了
df1.groupBy("Job").sum("sal").show
df1.groupBy("Job").max("sal").show
df1.groupBy("Job").min("sal").show
df1.groupBy("Job").avg("sal").show
df1.groupBy("Job").count.show
df1.groupBy("Job").avg("sal").where("avg(sal) > 2000").show
df1.groupBy("Job").avg("sal").where($"avg(sal)">2000).show
df1.groupBy("Job").agg("sal"->"max","sal"->"min","sal"->"avg","sal"->"sum","sal"->"count").show
df1.groupBy("deptno").agg("sal"->"max","sal"->"min","sal"->"avg","sal"->"sum","sal"->"count").show

orderBy相关

orderBy == sort

df1.orderBy("name").show(5)
df1.orderBy($"name".asc).show(5)
df1.orderBy(-$"age").show(5)

运行测试的结果如下图所示:
在这里插入图片描述
继续进行测试:

df1.sort("age").show(3)
df1.sort($"age".asc).show(3)
df1.sort(col("age")).show(3)

测试结果如下图所示:
在这里插入图片描述

JOIN相关

// 笛卡尔积
df1.crossJoin(df1).count
// 等值连接(单字段)
df1.join(df1,"name").count
// 等值连接(多字段)
df1.join(df1, Seq("name","age")).show

运行的测试结果如下图所示:
在这里插入图片描述

这里编写两个case:

// 第一个数据集caseclass StudentAge(sno:Int, name:String, age:Int)val lst = List(StudentAge(1,"Alice",18), StudentAge(2,"Andy",19), StudentAge(3,"Bob",17), StudentAge(4,"Justin",21), StudentAge(5,"Cindy",20))val ds1 = spark.createDataset(lst)// 第二个数据集caseclass StudentHeight(sname:String, height:Int)val rdd = sc.makeRDD(List(StudentHeight("Alice",160), StudentHeight("Andy",159), StudentHeight("Bob",170), StudentHeight("Cindy",165), StudentHeight("Rose",160)))val ds2 = rdd.toDS

运行测试的结果如下图所示:
在这里插入图片描述
接下来我们进行连表操作:

// 连表操作 不可以使用 "name"==="sname" !!!
ds1.join(ds2,'name==='sname).show
ds1.join(ds2, ds1("name")===ds2("sname")).show
ds1.join(ds2, $"name"===$"sname").show
ds1.join(ds2, $"name"===$"sname","inner").show

测试的运行结果如下图所示:
在这里插入图片描述

在这里插入图片描述

集合相关

val ds3 = ds1.select("name")val ds4 = ds2.select("sname")// union 求并集、不去重
ds3.union(ds4).show
// unionAll(过时了)与union等价// intersect 求交
ds3.intersect(ds4).show
// except 求差
ds3.except(ds4).show

运行结果如下图所示:
在这里插入图片描述

空值处理

math.sqrt(-1.0)
math.sqrt(-1.0).inNaN()
df1.show
// 删除所有列的空值和NaN
df1.na.drop.show
// 删除某列的空值和NaN
df1.na.drop(Array("xxx")).show
// 对列进行填充
df1.na.fill(1000).show
df1.na.fill(1000, Array("xxx")).show

本文转载自: https://blog.csdn.net/w776341482/article/details/141411696
版权归原作者 武子康 所有, 如有侵权,请联系我们删除。

“大数据-95 Spark 集群 SparkSQL Action与Transformation操作 详细解释与测试案例”的评论:

还没有评论