spark sql的简单操作

//从hdfs上读取数据文件并转为student对象进行操作
private def runSarkOnFile(spark:SparkSession){
import spark.implicits._
//读取数据文件 并生成rdd
var rdd=spark.read.textFile("hdfs://服务器ip:8020/tmp/dataTest/sparkStu.txt");
//对获取的rdd进行解析，并生成sutdent对象
var sturdd=rdd.map { x => x.split(" ")}.map { z => Student(z(0).toString(),z(1).toInt,z(2).toString())};
//显示student对象
sturdd.show();
//将sutdent对象注册成临时表 student
sturdd.registerTempTable("student");
//查询临时表中的数据，并显示
var sqlDF=spark.sql("select t.name,t.age,t.job from friend t where t.age>14 and t.age<26");
sqlDF.show();
}

private def applySchema(spark:SparkSession){
import spark.implicits._
import org.apache.spark.sql._
import org.apache.spark.sql.types._
//确定schema名称（列的名称）
var schemaString="name,age,job";
//解析schemaString,并生成StructType对象数组
var schemaType=StructType(schemaString.split(",").map { x => StructField(x,StringType,true)})
//从hdfs上读取数据文件
var stuDS=spark.sparkContext.textFile(path);
//使用Row对象，创建rowRdd
var sDS=stuDS.map { x => x.split(" ")}.map(s => Row(s(0),s(1),s(2)))
//创建schemaRDD
var rowDF=spark.createDataFrame(sDS, schemaType); // var rowDF=spark.sqlContext.applySchema(sDS, schemaType); 这种方法已经过时
 //打印schemaRDD的结构
rowDF.printSchema();
//注册Student table
rowDF.createOrReplaceTempView("Student"); // rowDF.registerTempTable("Student"); 这种方法已经过时
//rowDF.collect().foreach {print(_) }
//var resDS=spark.sql("select * from Student where age > 24");
var resDS=spark.sql("select name from Student");
resDS.show();
}

//使用parquet文件的方式
private def loadParquet(spark:SparkSession){
import spark.implicits._
//确定schema 列名称
var schemaString="name,age,job";
//解析schemaString,并生成StructType对象数组
var schemaType=StructType(schemaString.split(",").map { x => StructField(x,StringType,true)})
//创建rowRdd
var stuDS=spark.sparkContext.textFile(path);
var sDS=stuDS.map { x => x.split(" ")}.map(s => Row(s(0),s(1),s(2)))
//将schemaRDD保存成parquet文件
var rowDF=spark.sqlContext.applySchema(sDS, schemaType);
//将文件写到hdfs://服务器ip:8020/tmp/dataTest/
rowDF.write.parquet("hdfs://服务器ip:8020/tmp/dataTest/student.parquet");
-------------------------------------------------------------------
//读取parquet文件
var redParfile=spark.read.parquet("hdfs://服务器ip:8020/tmp/dataTest/student.parquet");
redParfile.createOrReplaceTempView("redParfilered");
var resultRdd=spark.sql("select * from redParfilered t where t.name=‘zhangxs‘");
//DataFrame.rdd 可以将dataFrame转为RDD类型
resultRdd.rdd.map { x => "name"+x(0) }.collect().foreach { print(_) }
}
 
/**
* spark可以自动的识别一个json模式并加载成数据集，
* 这种转换可以使用SparkSession.read.json() 函数
* 这个数据集的来源可以是一个rdd，也可以是一个json文件
*
*/
private def jsonFile(spark:SparkSession){
var jsonRdd=spark.read.json("hdfs://192.168.177.124:8020/tmp/dataTest/sparksql");
jsonRdd.createOrReplaceTempView("student");
var jfRdd= spark.sql("select * from student t where t.age >24");
jfRdd.show();

/**
* 使用Json类型的rdd加载json
*
* 如果加:: Nil，返回是一个char类型的rdd,加上则返回的是String类型的rdd
*/
var rdd=spark.sparkContext.makeRDD("""{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil);
var rddre=spark.read.json(rdd);
rddre.show();
}