大数据-95 Spark 集群 SparkSQL Action与Transformation操作 具体表明与测 ...

玛卡巴卡的卡巴卡玛  金牌会员 | 2024-8-24 14:12:10 | 来自手机 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 652|帖子 652|积分 1956

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

现在已经更新到了：



  • 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
天生数据

生存并上传到服务器上
  1. EMPNO,ENAME,JOB,MGR,HIREDATE,SAL,COMM,DEPTNO
  2. 7369,SMITH,CLERK,7902,2001-01-02 22:12:13,800,,20
  3. 7499,ALLEN,SALESMAN,7698,2002-01-02 22:12:13,1600,300,30
  4. 7521,WARD,SALESMAN,7698,2003-01-02 22:12:13,1250,500,30
  5. 7566,JONES,MANAGER,7839,2004-01-02 22:12:13,2975,,20
  6. 7654,MARTIN,SALESMAN,7698,2005-01-02 22:12:13,1250,1400,30
  7. 7698,BLAKE,MANAGER,7839,2005-04-02 22:12:13,2850,,30
  8. 7782,CLARK,MANAGER,7839,2006-03-02 22:12:13,2450,,10
  9. 7788,SCOTT,ANALYST,7566,2007-03-02 22:12:13,3000,,20
  10. 7839,KING,PRESIDENT,,2006-03-02 22:12:13,5000,,10
  11. 7844,TURNER,SALESMAN,7698,2009-07-02 22:12:13,1500,0,30
  12. 7876,ADAMS,CLERK,7788,2010-05-02 22:12:13,1100,,20
  13. 7900,JAMES,CLERK,7698,2011-06-02 22:12:13,950,,30
  14. 7902,FORD,ANALYST,7566,2011-07-02 22:12:13,3000,,20
  15. 7934,MILLER,CLERK,7782,2012-11-02 22:12:13,1300,,10
复制代码
写入内容如下图所示：

测试运行

我们进入 spark-shell 举行测试
  1. // 处理头，使用自动类型推断
  2. val df1 = spark.read.option("header", true).option("infershema", "true").csv("test_spark_03.txt")
  4. df1.count
  5. // 缺省显示20行
  6. df1.union(df1).show()
  7. // 显示2行
  8. df1.show(2)
复制代码
执行结果如下图所示：

继承举行测试：
  1. // 不截断字符
  2. df1.toJSON.show(false)
  3. // 显示10行 不截断字符
  4. df1.toJSON.show(10, false)
复制代码
运行结果如下图所示：

继承举行测试：
  1. // collect 返回数组 Array[Row]
  2. val c1 = df1.collect()
  3. // collectAsList 返回List Lits[Row]
  4. val c2 = df1.collectAsList()
  6. // 返回 Row
  7. val h1 = df1.head()
  8. val f1 = df1.first()
  10. // 返回 Array[Row]
  11. val h2 = df1.head(3)
  12. val f2 = df1.take(3)
  14. // 返回 List[Row]
  15. val t2 = df1.takeAsList(2)
复制代码
运行结果如下图所示：

继承举行测试：
  1. // 结构属性
  2. // 查看列名
  3. df1.columns
  4. // 查看列名和类型
  5. df1.dtypes
  6. // 查看执行计划
  7. df1.explain()
  8. // 获取某个列
  9. df1.col("ENAME")
  10. // 常用
  11. df1.printSchema
复制代码
运行结果如下图所示：

Transformation 操作



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



  • map
  • filter
  • flatMap
  • mapPartitions
  • sample
  • randomSplit
  • limt
  • distinct
  • dropDuplicates
  • describe
我们举行测试：
  1. val df1 = spark.read.csv("/opt/wzk/data/people1.csv")
  2. // 获取第1列
  3. df1.map(row => row.getAs[String](0)).show
  5. // randomSplit 将DF、DS按给定参数分成多份
  6. val df2 = df1.randomSplit(Array(0.5, 0.6, 0.7))
  7. df2(0).count
  8. df2(1).count
  9. df2(2).count
复制代码
测试结果如下图：

我们继承举行测试：
  1. // 取10行数据生成新的Dataset
  2. val df2 = df1.limit(10)
  3. // distinct 去重
  4. val df2 = df1.union(df1)
  5. df2.distinct.count
  7. // dropDuplicates 按列值去重
  8. df2.dropDuplicates.show
  9. df2.dropDuplicates("_c0").show
复制代码
执行结果如下图：

存储相关



  • cacheTable
  • persist
  • checkpoint
  • unpersist
  • cache
备注：Dataset默认的存储级别是 MEMEORY_AND_DISK
  1. spark.sparkContext.setCheckpointDir("hdfs://h121.wzk.icu:9000/checkpoint")
  3. df1.show()
  4. df1.checkpoint()
  5. df1.cache()
  7. import org.apache.spark.storage.StorageLevel
  8. df1.persist(StorageLevel.MEMORY_ONLY)
  9. df1.count()
  10. df1.unpersist(true)
复制代码
执行结果如下图所示：

select相关



  • 列的多种表现
  • select
  • selectExpr
启动 Spark-Shell 继承举行测试
  1. // 这里注意 option("header", "true") 自动解析一下表头
  2. val df1 = spark.read.option("header", "true").csv("/opt/wzk/data/people1.csv")
  4. // $ col() 等等 不可以混用！！！（有解决方法，但是建议不混用！！！）
  5. // 可以多种形式获取到列
  6. df1.select($"name", $"age", $"job").show
复制代码
执行结果如下图所示：

继承举行测试

  1. df1.select("name", "age", "job").show(3)
  2. df1.select(col("name"), col("age"), col("job")).show(3)
  3. df1.select($"name", $"age"+1000, $"job").show(5)
复制代码
运行结果如下图所示：

where相关

接着对上述内容举行测试：
  1. df1.filter("age > 25").show
  2. df1.filter("age > 25 and name == 'wzk18'").show
  3. df1.where("age > 25").show
  4. df1.where("age > 25 and name == 'wzk19'").show
复制代码
运行测试结果如下图：

groupBy相关



  • groupBy
  • agg
  • max
  • min
  • avg
  • sum
  • count
举行测试：
  1. // 由于我的字段中没有数值类型的，就不做测试了
  2. df1.groupBy("Job").sum("sal").show
  3. df1.groupBy("Job").max("sal").show
  4. df1.groupBy("Job").min("sal").show
  5. df1.groupBy("Job").avg("sal").show
  6. df1.groupBy("Job").count.show
  7. df1.groupBy("Job").avg("sal").where("avg(sal) > 2000").show
  8. df1.groupBy("Job").avg("sal").where($"avg(sal)" > 2000).show
  9. df1.groupBy("Job").agg("sal"->"max", "sal"->"min", "sal"-
  10. >"avg", "sal"->"sum", "sal"->"count").show
  11. df1.groupBy("deptno").agg("sal"->"max", "sal"->"min", "sal"-
  12. >"avg", "sal"->"sum", "sal"->"count").show
复制代码
orderBy相关

orderBy == sort
  1. df1.orderBy("name").show(5)
  2. df1.orderBy($"name".asc).show(5)
  3. df1.orderBy(-$"age").show(5)
复制代码
运行测试的结果如下图所示：

继承举行测试：
  1. df1.sort("age").show(3)
  2. df1.sort($"age".asc).show(3)
  3. df1.sort(col("age")).show(3)
复制代码
测试结果如下图所示：

JOIN相关

  1. // 笛卡尔积
  2. df1.crossJoin(df1).count
  3. // 等值连接（单字段）
  4. df1.join(df1, "name").count
  5. // 等值连接（多字段）
  6. df1.join(df1, Seq("name", "age")).show
复制代码
运行的测试结果如下图所示：

这里编写两个case：
  2. // 第一个数据集
  3. case class StudentAge(sno: Int, name: String, age: Int)
  5. val lst = List(StudentAge(1,"Alice", 18), StudentAge(2,"Andy", 19), StudentAge(3,"Bob", 17), StudentAge(4,"Justin", 21), StudentAge(5,"Cindy", 20))
  7. val ds1 = spark.createDataset(lst)
  9. // 第二个数据集
  10. case class StudentHeight(sname: String, height: Int)
  12. val rdd = sc.makeRDD(List(StudentHeight("Alice", 160), StudentHeight("Andy", 159), StudentHeight("Bob", 170), StudentHeight("Cindy", 165), StudentHeight("Rose", 160)))
  14. val ds2 = rdd.toDS
复制代码
运行测试的结果如下图所示：

接下来我们举行连表操作：
  1. // 连表操作 不可以使用 "name"==="sname" !!!
  2. ds1.join(ds2, 'name==='sname).show
  3. ds1.join(ds2, ds1("name")===ds2("sname")).show
  4. ds1.join(ds2, $"name"===$"sname").show
  5. ds1.join(ds2, $"name"===$"sname", "inner").show
复制代码
测试的运行结果如下图所示：


聚集相关

  1. val ds3 = ds1.select("name")
  2. val ds4 = ds2.select("sname")
  4. // union 求并集、不去重
  5. ds3.union(ds4).show
  6. // unionAll(过时了)与union等价
  7. // intersect 求交
  8. ds3.intersect(ds4).show
  9. // except 求差
  10. ds3.except(ds4).show
复制代码
运行结果如下图所示：

空值处理

  1. math.sqrt(-1.0)
  2. math.sqrt(-1.0).inNaN()
  3. df1.show
  4. // 删除所有列的空值和NaN
  5. df1.na.drop.show
  6. // 删除某列的空值和NaN
  7. df1.na.drop(Array("xxx")).show
  8. // 对列进行填充
  9. df1.na.fill(1000).show
  10. df1.na.fill(1000, Array("xxx")).show
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

玛卡巴卡的卡巴卡玛

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表