Spark笔记(三) DataFrame

叁歲伎倆 2021-12-11 03:53 247阅读 0赞

import org.apache.spark.sql.SparkSession
    
    
    object test24 { def main(args: Array[String]): Unit = { val ss= SparkSession.builder().master("local[2]").appName("hello").getOrCreate() val sc=ss.sparkContext ss.sparkContext.setLogLevel("ERROR") val df =ss.read.json("people.json").select("name","age","height","address","sex","phone") //df打印 df.show(10,false) // numRows表示展示行数,Truncate表示是否最多显示20个字符 //df统计 总条数,平均值,标准差,最大值,最小值 df.describe("age","height").show() //describe 获取指定字段的统计信息 //df转成数组 val df_arr=df.collect() println(df_arr.toBuffer) //df转成list val df_list=df.collectAsList() println(df_list) //获取若干行记录 println(df.first()+" , "+df.head()+" , "+df.head(2)+" , "+df.take(3)+" , "+df.takeAsList(4)) //where 条件相关 df.where("name='apple' or age>=23").show() //过滤 df.filter("name !='orange' and age=23").show() //查询指定字段 df.select("name","age").show() //重载select方法 df.select(df("name"),df("age"),df("age")+1).show() //对列重新命名 df.selectExpr("name","age as nianling","round(age)").show() //col: 获取指定字段,返回值Column类型 val nameCol=df.col("name") //apply: 获取指定列,返回值Column类型 val ageCol1=df.apply("age") val ageCol2=df("age") //等价上一个 //drop: 去除指定字段,保留其他字段 df.drop("name")// def drop(colName:String) 传入String类型的列名称 df.drop(df("name")) // def drop(col:Column) 传入Column类型的列 //排序 df.orderBy(df("age")).show()//默认升序,或者asc df.orderBy(df("age").desc).show()//降序 df.orderBy(df("age").desc,df("name")).show()//多字段排序,age降序,name升序 df.sort(df("age").desc,df("name")).show()//同 orderBy //groupBy()后得到RelationalGroupedDataset类型对象,这个后续api操作有: // Count(分组中元素个数),max(最大值),min(最小值),mean(平均值),sum(总和) df.groupBy("age").count().show() df.groupBy("address","sex").mean("height").show() //distinct去除重复行 df.distinct().show() //dropDuplicates 指定字段去重 df.dropDuplicates("name").show() df.dropDuplicates("age","address").show() //聚合操作agg方法 df.agg("age"->"mean","height"->"max").show()//对age求平均值,对height求最大值 df.groupBy("address").agg("age"->"mean","height"->"max") //聚合操作一般与groupBy方法配合使用 //union:对字段一致的两个DataFrame组合 val df2=df.limit(1) df.union(df2).show() //join val joinDf1=df.select("name","age","height","phone") val joinDf2=df.select("name","address","sex","phone") //单字段join joinDf1.join(joinDf2,"name").show() //按name内连接 //多字段join joinDf1.join(joinDf2,Seq("name","phone")) //指定join类型:inner,left_outer,right_outer,leftsemiinner,cross joinDf1.join(joinDf2,Seq("name","phone"),"right_outer") //获取指定字段统计信息 //stat方法:用于计算指定字段之间的统计信息,比如方差,协方差等,返回DataFrameStatFunctions //freqItems用于计算一列或几列中出现频繁的的值的集合 df.stat.freqItems(Array("age","height"),0.3).show() //统计age,height字段出现频率在30%以上的内容 df.stat.freqItems(Seq("age","height"),0.3).show() //同上, 第一个参数是Array或者Seq, 第二个参数是频率 //corr 求两列相关性,返回一个Double df.stat.corr("age","height") //计算age和height的相关性 //cov 求两列的协方差 df.stat.cov("age","height") //获取两个DataFram的交集 val df3=df.limit(2) df.intersect(df3).show() //获取df中去掉df3后的记录 df.except(df3).show() //处理空值列 df.na.drop() //只要有空值的列都删除掉 df.na.drop(Array("sex","phone")).show()//指定sex,phone列中含有空值的删除 df.na.fill(Map(("age",-1),("address","CN"))).show() //指定用-1来替换age为空的值,用CN来替换address为空的值 } }

![1374675-20190530191911792-1176993142.png][]

转载于:https://www.cnblogs.com/ShyPeanut/p/10932802.html

[1374675-20190530191911792-1176993142.png]: /images/20211211/018001117ed944d391d71d9573f4c8af.png