SparkSQL如何运用

今天小编给大家分享一下SparkSQL如何运用的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。

一：SparkSQL

1.SparkSQL简介

Spark SQL是Spark的一个模块，用于处理结构化的数据，它提供了一个数据抽象DataFrame（最核心的编程抽象就是DataFrame），并且SparkSQL作为分布式SQL查询引擎。
Spark SQL就是将SQL转换成一个任务，提交到集群上运行，类似于Hive的执行方式。

2.SparkSQL运行原理

将Spark SQL转化为RDD，然后提交到集群执行。

3.SparkSQL特点

（1）容易整合，Spark SQL已经集成在Spark中

（2）提供了统一的数据访问方式：JSON、CSV、JDBC、Parquet等都是使用统一的方式进行访问

（3）兼容 Hive

（4）标准的数据连接：JDBC、ODBC

二、SparkSQL运用

package sqlimport org.apache.avro.ipc.specific.Personimport org.apache.sparkimport org.apache.spark.rdd.RDDimport org.apache.spark.sqlimport org.apache.spark.sql.catalyst.InternalRowimport org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}import org.junit.Testclass Intro {  @Test  def dsIntro(): Unit ={    val spark: SparkSession = new sql.SparkSession.Builder()      .appName("ds intro")      .master("local[6]")      .getOrCreate()    //导入隐算是shi转换    import spark.implicits._    val sourceRDD: RDD[Person] =spark.sparkContext.parallelize(Seq(Person("张三",10),Person("李四",15)))    val personDS: Dataset[Person] =sourceRDD.toDS();//personDS.printSchema()打印出错信息    val resultDS: Dataset[Person] =personDS.where('age>10)      .select('name,'age)      .as[Person]    resultDS.show()  }  @Test  def dfIntro(): Unit ={    val spark: SparkSession =new SparkSession.Builder()      .appName("ds intro")      .master("local")      .getOrCreate()    import spark.implicits._    val sourceRDD: RDD[Person] = spark.sparkContext.parallelize(Seq(Person("张三",10),Person("李四",15)))    val df: DataFrame = sourceRDD.toDF()//隐shi转换    df.createOrReplaceTempView("person")//创建表    val resultDF: DataFrame =spark.sql("select name from person where age>=10 and age<=20")    resultDF.show()  }  @Test  def database1(): Unit ={    //1.创建sparkSession    val spark: SparkSession =new SparkSession.Builder()      .appName("database1")      .master("local[6]")      .getOrCreate()      //2.导入引入shi子转换    import spark.implicits._    //3.演示    val sourceRDD: RDD[Person] =spark.sparkContext.parallelize(Seq(Person("张三",10),Person("李四",15)))    val dataset: Dataset[Person] =sourceRDD.toDS()    //Dataset 支持强类型的API    dataset.filter(item => item.age >10).show()    //Dataset 支持若弱类型的API    dataset.filter('age>10).show()    //Dataset 可以直接编写SQL表达式    dataset.filter("age>10").show()  }  @Test  def database2(): Unit ={    val spark: SparkSession = new SparkSession.Builder()      .master("local[6]")      .appName("database2")      .getOrCreate()    import spark.implicits._    val dataset: Dataset[Person] =spark.createDataset(Seq(Person("张三",10),Person("李四",20)))    //无论Dataset中放置的是什么类型的对象,最终执行计划中的RDD上都是internalRow    //直接获取到已经分析和解析过得Dataset的执行计划,从中拿到RDD    val executionRdd: RDD[InternalRow] =dataset.queryExecution.toRdd    //通过将Dataset底层的RDD通过Decoder转成了和Dataset一样的类型RDD    val typedRdd:RDD[Person] = dataset.rdd    println(executionRdd.toDebugString)    println()    println()    println(typedRdd.toDebugString)  }  @Test  def database3(): Unit = {    //1.创建sparkSession    val spark: SparkSession = new SparkSession.Builder()      .appName("database1")      .master("local[6]")      .getOrCreate()    //2.导入引入shi子转换    import spark.implicits._    val dataFrame: DataFrame = Seq(Person("zhangsan", 15), Person("lisi", 20)).toDF()    //3.看看DataFrame可以玩出什么花样    //select name from...    dataFrame.where('age > 10)      .select('name)      .show()  }//  @Test//  def database4(): Unit = {//    //1.创建sparkSession//    val spark: SparkSession = new SparkSession.Builder()//      .appName("database1")//      .master("local[6]")//      .getOrCreate()//    //2.导入引入shi子转换//    import spark.implicits._//    val personList=Seq(Person("zhangsan",15),Person("lisi",20))////    //1.toDF//    val df1: DataFrame =personList.toDF()//    val df2: DataFrame =spark.sparkContext.parallelize(personList).toDF()//      //2.createDataFrame//    val df3: DataFrame =spark.createDataFrame(personList)////    //3.read//    val df4: DataFrame =spark.read.csv("")//    df4.show()//  }  //toDF()是转成DataFrame,toDs是转成Dataset  //  DataFrame就是Dataset[Row] 代表弱类型的操作,Dataset代表强类型的操作,中的类型永远是row,DataFrame可以做到运行时类型安全,Dataset可以做到 编译时和运行时都安全@Testdef database4(): Unit = {  //1.创建sparkSession  val spark: SparkSession = new SparkSession.Builder()    .appName("database1")    .master("local[6]")    .getOrCreate()  //2.导入引入shi子转换  import spark.implicits._  val personList=Seq(Person("zhangsan",15),Person("lisi",20))  //DataFrame代表弱类型操作是编译时不安全  val df: DataFrame =personList.toDF()  //Dataset是强类型的  val ds: Dataset[Person] =personList.toDS()  ds.map((person:Person) =>Person(person.name,person.age))}  @Test  def row(): Unit ={    //1.Row如何创建,它是什么    //row对象必须配合Schema对象才会有列名    val p: Person =Person("zhangsan",15)    val row: Row =Row("zhangsan",15)    //2.如何从row中获取数据    row.getString(0)    row.getInt(1)    //3.Row也是样例类、    row match {      case Row(name,age) => println(name,age)    }  }}case class Person(name: String, age: Int)

以上就是"SparkSQL如何运用"这篇文章的所有内容，感谢各位的阅读！相信大家阅读完这篇文章都有很大的收获，小编每天都会为大家更新不同的知识，如果还想学习更多的知识，请关注行业资讯频道。