2023学年完整公开课版DataFrame与SparkSQL编程

DataFrame与SparkSQL编程DataFrame与SparkSQLnameage第1步：制作“表头”第2步：制作“表中的记录”“Grace”29“Andy”30“John”19第3步：将表头和表中的记录拼装在一起nameage“Grace”29“Andy”30“John”19DataFrame与SparkSQL1导入相应的包scala>importorg.apache.spark.sql.types._importorg.apache.spark.sql.types._

scala>importorg.apache.spark.sql.Rowimportorg.apache.spark.sql.Row

DataFrame与SparkSQL2加载文件生成RDD//生成RDDscala>valpeopleRDD=spark.sparkContext.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt")DataFrame与SparkSQL3制作表头

scala>valschemaString="nameage"

//根据模式字符串生成模式scala>valfields=schemaString.split("").map(fieldName=>StructField(fieldName,StringType,nullable=true))scala>valschema=StructType(fields)//从上面信息可以看出，schema描述了模式信息，模式中包含name和age两个字段

DataFrame与SparkSQL4对peopleRDD的每一行元素进行解析，将表头和表记录拼装起来scala>valrowRDD=peopleRDD.map(_.split(",")).map(attributes=>Row(attributes(0),attributes(1).trim))rowRDD:org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]=MapPartitionsRDD[3]atmapat<console>:29

scala>valpeopleDF=spark.createDataFrame(rowRDD,schema)DataFrame与SparkSQL5注册为临时表供SQL查询使用//必须注册为临时表才能供下面查询使用scala>peopleDF.createOrReplaceTempView("people")scala>valresults=spark.sql("SELECTname,ageFROMpeople")results:org.apache.spark.sql.DataFrame=[name:string,age:string]

scala>results.map(attributes=>"name:"+attributes(0)+","+"age:"+attributes(1)).show()DataFrame与SparkSQL6显示查询结果一个小例子1导入相应的包importorg.apache.spark.sql.SparkSessionimportorg.apache.spark.sql.types.{IntegerType,StringType,StructField,StructType}一个小例子2制作表头valspark=SparkSession.builder().master("local").appName("dateprocess").getOrCreate()valschema=StructType(List(StructField("user_id",IntegerType),StructField("item_id",IntegerType),StructField("category_id",IntegerType),StructField("behavior",StringType)))一个小例子3生成DataFramevaldf=spark.read.format("com.databricks.spark.csv").schema(schema).option("header","false").load("hdfs://localhost:9000/dbtaobao/dataset/user_behavior/UserBehavior.csv")一个小例子4注册为临时表供SQL查询使用//必须注册为临时表才能供下面查询使用df.createOrReplaceTempView("user_behavior")//统计页面浏览量valbehavior_df=spark.sql("selectuser_id,count(if(beha