https://www.bilibili.com/video/BV1Xz4y1m7cv?p=59
SparkSQL花式查询在Spark5QL模块中,将结构化数据封装到Dataframe或Dataset集合中后,提供了两种方式分析处理数据:
1、SQL编程,将Dataframe/Dataset注册为临时视图或表,编写SQL语句,类似HiveQL;
2、DSL (domain-specific language)编程,(类似于面向对象)调用Dataframe/Dataset APIl(函数),类似RDD中函数;
针对personDF中的数据使用SQL和DSL两种方式进行各种查询
代码实现:package sql import org.apache.spark.rdd.RDD import org.apache.spark.sql.SparkSession object Demo04_Query { def main(args: Array[String]): Unit = { //TODO 0.准备环境 val spark = SparkSession.builder().appName("sparksql").master("local[*]").getOrCreate() val sc = spark.sparkContext sc.setLogLevel("WARN") //TODO 1.加载数据 val lines:RDD[String] = sc.textFile("data/SogouE.txt") //TODO 2.处理数据 val personRDD:RDD[Person] = lines.map(line => { val arr: Array[String] = line.split(" ") Person(arr(0).toInt, arr(1), arr(2).toInt) }) //RDD--->DF import spark.implicits._ val personDF = personRDD.toDF() //TODO=========SQL=========== //注册表名 // personDF.registerTempTable("")//过期的 // personDF.createOrReplaceGlobalTempView("")//创建全局的,SparkSession也可以用,但是生命周期太长! personDF.createOrReplaceTempView("t_person")//创建临时的,当前SparkSession也可以用 //=1.查看name字段的数据 spark.sql("select name from t_person").show() //=2.查看name和age字段数据 spark.sql("select name,age from t_person").show() //=3.查看所有的name和age,并将age+1 spark.sql("select name,age,age+1 from t_person").show() //=4.过滤age大于等于25的 spark.sql("select name,age from t_person where age>=25").show() //=5.统计年龄大于30的人数 spark.sql("select count(*) from t_person where age>30").show() //=6.按年龄进行分组并统计相同年龄的人数 spark.sql("select age,count(*) from t_person where group by age").show() //=7.查询姓名=张三的 spark.sql("select name from t_person wheree name='zhangsan'").show() //TODO=========DSL:面向对象的SQL=========== //=1.查看name字段的数据 // personDF.select(personDF.col("name")) personDF.select("name").show() //=2.查看name和age字段数据 personDF.select("name","age").show() //=3.查看所有的name和age,并将age+1 // personDF.select("name","age","age+1").show() //注意是$把列名转为了对象 personDF.select($"name",$"age",$"age+1"+1).show() //注意是$把字符串转为了对象 personDF.select('name,'age,'age+1).show() //=4.过滤age大于等于25的 personDF.filter("age>=25").show() personDF.filter($"age">=25).show() personDF.filter('age>=25).show() //=5.统计年龄大于30的人数 val count: Long = personDF.where('age>30).count() println("年龄大于30的人数 为:"+count) //=6.按年龄进行分组并统计相同年龄的人数 personDF.groupBy('age).count().show() //=7.查询姓名=张三的 personDF.filter("name='zhangsan'").show() personDF.filter($"name"==="zhangsan").show() personDF.filter('name ==="zhangsan").show() //TODO 3.输出数据 personDF.printSchema() personDF.show() //TODO 4.关闭资源 spark.close() } case class Person(id:Int,name:String,age:Int) }补充
1.查看name字段的数据
2.查看name和age字段数据
3.查看所有的name和age,并将age+1
4.过滤age大于等于25的
5.统计年龄大于30的人数
6.按年龄进行分组并统计相同年龄的人数
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)