SparkSQL语法及API
一、SparkSql基础语法
1、通过方法来使用
1.查询
df.select("id","name").show();
- 1
1>带条件的查询
df.select($"id",$"name").where($"name" === "bbb").show()
- 1
2>排序查询
orderBy/sort($"列名") 升序排列
orderBy/sort($"列名".desc) 降序排列
orderBy/sort($"列1" , $"列2".desc) 按两列排序
例如:
df.select($"id",$"name").orderBy($"name".desc).show
df.select($"id",$"name").sort($"name".desc).show
tabx.select($"id",$"name").sort($"id",$"name".desc).show
- 1
- 2
- 3
3>分组查询
groupBy("列名", ...).max(列名) 求最大值
groupBy("列名", ...).min(列名) 求最小值
groupBy("列名", ...).avg(列名) 求平均值
groupBy("列名", ...).sum(列名) 求和
groupBy("列名", ...).count() 求个数
groupBy("列名", ...).agg 可以将多个方法进行聚合
例如:
scala>val rdd = sc.makeRDD(List((1,"a","bj",100),(2,"b","sh",80),(3,"c","gz",50),(4,"d","bj",45),(5,"e","gz",90)));
scala>val df = rdd.toDF("id","name","addr","score");
scala>df.groupBy("addr").count().show()
scala>df.groupBy("addr").agg(max($"score"), min($"score"), count($"*")).show
- 1
- 2
- 3
- 4
4>连接查询
scala>val dept=sc.parallelize(List((100,"caiwubu"),(200,"yanfabu"))).toDF("deptid","deptname")
scala>val emp=sc.parallelize(List((1,100,"zhang"),(2,200,"li"),(3,300,"wang"))).toDF("id","did","name")
scala>dept.join(emp,$"deptid" === $"did").show
scala>dept.join(emp,$"deptid" === $"did","left").show
- 1
- 2
- 3
- 4
左向外联接的结果集包括 LEFT OUTER子句中指定的左表的所有行,而不仅仅是联接列所匹配的行。如果左表的某行在右表中没有匹配行,则在相关联的结果集行中右表的所有选择列表列均为空值。
scala>dept.join(emp,$"deptid" === $"did","right").show
- 1
2.执行运算
val df = sc.makeRDD(List(1,2,3,4,5)).toDF("num");
df.select($"num" * 100).show
- 1
- 2
3.使用列表
val df = sc.makeRDD(List(("zhang",Array("bj","sh")),("li",Array("sz","gz")))).toDF("name","addrs")
df.selectExpr("name","addrs[0]").show
- 1
- 2
使用结构体:
{"name":"王二小","address":{"city":"大土坡","street":"南二环甲字1号"}}
{"name":"流放","address":{"city":"天涯海角","street":"南二环甲字2号"}}
- 1
- 2
val df = sqlContext.read.json("file:///root/work/users.json")
dfs.select("name","address.street").show
- 1
- 2
其他
df.count//获取记录总数
val row = df.first()//获取第一条记录
val value = row.getString(1)//获取该行指定列的值
df.collect //获取当前df对象中的所有数据为一个Array 其实就是调用了df对象对应的底层的rdd的collect方法
- 1
- 2
- 3
- 4
2、通过sql语句来调用
1.针对表的操作
1>创建表
df.registerTempTable("tabName")
- 1
2>查看表
sqlContext.sql("show tables").show
- 1
2.查询
val sqc = new org.apache.spark.sql.SQLContext(sc);
val df = sc.makeRDD(List((1,"a","bj"),(2,"b","sh"),(3,"c","gz"),(4,"d","bj"),(5,"e","gz"))).toDF("id","name","addr");
df.registerTempTable("stu");
sqc.sql("select * from stu").show()
- 1
- 2
- 3
- 4
1>带条件的查询
val df = sc.makeRDD(List((1,"a","bj"),(2,"b","sh"),(3,"c","gz"),(4,"d","bj"),(5,"e","gz"))).toDF("id","name","addr");
df.registerTempTable("stu");
sqc.sql("select * from stu where addr = 'bj'").show()
- 1
- 2
- 3
2>排序查询
val sqlContext = new org.apache.spark.sql.SQLContext(sc);
val df = sc.makeRDD(List((1,"a","bj"),(2,"b","sh"),(3,"c","gz"),(4,"d","bj"),(5,"e","gz"))).toDF("id","name","addr");
df.registerTempTable("stu");
sqlContext.sql("select * from stu order by addr").show()
sqlContext.sql("select * from stu order by addr desc").show()
- 1
- 2
- 3
- 4
- 5
3>分组查询
val sqlContext = new org.apache.spark.sql.SQLContext(sc);
val df = sc.makeRDD(List((1,"a","bj"),(2,"b","sh"),(3,"c","gz"),(4,"d","bj"),(5,"e","gz"))).toDF("id","name","addr");
df.registerTempTable("stu");
sqlContext.sql("select addr,count(*) from stu group by addr").show()
- 1
- 2
- 3
- 4
4>连接查询
val sqlContext = new org.apache.spark.sql.SQLContext(sc);
val dept=sc.parallelize(List((100,"财务部"),(200,"研发部"))).toDF("deptid","deptname")
val emp=sc.parallelize(List((1,100,"张财务"),(2,100,"李会计"),(3,300,"王研发"))).toDF("id","did","name")
dept.registerTempTable("deptTab");
emp.registerTempTable("empTab");
sqlContext.sql("select deptname,name from deptTab inner join empTab on deptTab.deptid = empTab.did").show()
- 1
- 2
- 3
- 4
- 5
- 6
5>分页查询
val sqlContext = new org.apache.spark.sql.SQLContext(sc);
val df = sc.makeRDD(List(1,2,3,4,5)).toDF("num");
df.registerTempTable("tabx")
sqlContext.sql("select * from tabx limit 3").show();
- 1
- 2
- 3
- 4
3.执行运算
val sqlContext = new org.apache.spark.sql.SQLContext(sc);
val df = sc.makeRDD(List(1,2,3,4,5)).toDF("num");
df.registerTempTable("tabx")
sqlContext.sql("select num * 100 from tabx").show();
- 1
- 2
- 3
- 4
4.类似hive方式的操作
scala>val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
scala>hiveContext.sql("create table if not exists zzz (key int, value string) row format delimited fields terminated by '|'")
scala>hiveContext.sql("load data local inpath 'file:///home/software/hdata.txt' into table zzz")
scala>hiveContext.sql("select key,value from zzz").show
- 1
- 2
- 3
- 4
5.案例
val sqlContext = new org.apache.spark.sql.SQLContext(sc);
val df = sc.textFile("file:///root/work/words.txt").flatMap{ _.split(" ") }.toDF("word")
df.registerTempTable("wordTab")
sqlContext.sql("select word,count(*) from wordTab group by word").show
- 1
- 2
- 3
- 4
二、SparkSql API
可以通过java API使用sparksql。
1、创建工程
打开scala IDE开发环境,创建一个scala工程。
2、导入jar包
导入spark相关依赖jar包。
3、创建类
创建包路径以object类。