创建DataFrame的DSL和SQL方式分别如何操作的
Admin 2022-08-05 群英技术资讯 331 次浏览
7369,SMITH,CLERK,7902,1980/12/17,800,20
7499,ALLEN,SALESMAN,7698,1981/2/20,1600,300,30
7521,WARD,SALESMAN,7698,1981/2/22,1250,500,30
7566,JONES,MANAGER,7839,1981/4/2,2975,20
7654,MARTIN,SALESMAN,7698,1981/9/28,1250,1400,30
7698,BLAKE,MANAGER,7839,1981/5/1,2850,30
7782,CLARK,MANAGER,7839,1981/6/9,2450,10
7788,SCOTT,ANALYST,7566,1987/4/19,3000,20
7839,KING,PRESIDENT,1981/11/17,5000,10
7844,TURNER,SALESMAN,7698,1981/9/8,1500,0,30
7876,ADAMS,CLERK,7788,1987/5/23,1100,20
7900,JAMES,CLERK,7698,1981/12/3,9500,30
7902,FORD,ANALYST,7566,1981/12/3,3000,20
7934,MILLER,CLERK,7782,1982/1/23,1300,10
package com.scala.demo.sql import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.types.{DataType, DataTypes, StructField, StructType} object Demo01 { def main(args: Array[String]): Unit = { // 1.创建SparkContext和SparkSession对象 val sc = new SparkContext(new SparkConf().setAppName("Demo01").setMaster("local[2]")) val sparkSession = SparkSession.builder().getOrCreate() // 2. 使用StructType来定义Schema val mySchema = StructType(List( StructField("empno", DataTypes.IntegerType, false), StructField("ename", DataTypes.StringType, false), StructField("job", DataTypes.StringType, false), StructField("mgr", DataTypes.StringType, false), StructField("hiredate", DataTypes.StringType, false), StructField("sal", DataTypes.IntegerType, false), StructField("comm", DataTypes.StringType, false), StructField("deptno", DataTypes.IntegerType, false) )) // 3. 读取数据 val empRDD = sc.textFile("file:///D:\\TestDatas\\emp.csv") // 4. 将其映射成ROW对象 val rowRDD = empRDD.map(line => { val strings = line.split(",") Row(strings(0).toInt, strings(1), strings(2), strings(3), strings(4), strings(5).toInt,strings(6), strings(7).toInt) }) // 5. 创建DataFrame val dataFrame = sparkSession.createDataFrame(rowRDD, mySchema) // 6. 展示内容 DSL dataFrame.groupBy("deptno").sum("sal").as("result").sort("sum(sal)").show() } }
结果如下:
package com.scala.demo.sql import org.apache.spark.rdd.RDD import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.sql.types.{DataType, DataTypes, StructField, StructType} // 0. 数据分析 // 7499,ALLEN,SALESMAN,7698,1981/2/20,1600,300,30 // 1. 定义Emp样例类 case class Emp(empNo:Int,empName:String,job:String,mgr:String,hiredate:String,sal:Int,comm:String,deptNo:Int) object Demo02 { def main(args: Array[String]): Unit = { // 2. 读取数据将其映射成Row对象 val sc = new SparkContext(new SparkConf().setMaster("local[2]").setAppName("Demo02")) val mapRdd = sc.textFile("file:///D:\\TestDatas\\emp.csv") .map(_.split(",")) val rowRDD:RDD[Emp] = mapRdd.map(line => Emp(line(0).toInt, line(1), line(2), line(3), line(4), line(5).toInt, line(6), line(7).toInt)) // 3。创建dataframe val spark = SparkSession.builder().getOrCreate() // 引入spark隐式转换函数 import spark.implicits._ // 将RDD转成Dataframe val dataFrame = rowRDD.toDF // 4.2 sql语句操作 // 1、将dataframe注册成一张临时表 dataFrame.createOrReplaceTempView("emp") // 2. 编写sql语句进行操作 spark.sql("select deptNo,sum(sal) as total from emp group by deptNo order by total desc").show() // 关闭资源 spark.stop() sc.stop() } }
结果如下:
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:mmqy2019@163.com进行举报,并提供相关证据,查实之后,将立刻删除涉嫌侵权内容。
猜你喜欢
SQL 中运算符有很多,与IN、ANY、ALL等运算符不同,EXISTS运算符是单目运算法,可以用来判断查询字句是否有记录。这篇文章就主要介绍EXISTS 运算符的使用,感兴趣的朋友继续往下看吧。
这篇文章主要为大家详细介绍了数据库优化方案,具有一定的参考价值和实用性,感兴趣的小伙伴们可以参考一下
这篇文章主要介绍了SQL Server中避免触发镜像SUSPEND的N种方法,需要的朋友可以参考下
本文接下来讲解的是MySQL的隐式转换,对此感兴趣的小伙伴很可以一起来看看
这篇文章主要介绍了SQL SERVER提交事务回滚机制,本篇文章通过简要的案例,讲解了该项技术的了解与使用,以下就是详细内容,需要的朋友可以参考下
成为群英会员,开启智能安全云计算之旅
立即注册Copyright © QY Network Company Ltd. All Rights Reserved. 2003-2020 群英 版权所有
增值电信经营许可证 : B1.B2-20140078 粤ICP备09006778号 域名注册商资质 粤 D3.1-20240008