Spark之【SparkSQL编程】系列(No4)——《IDEA创建SparkSQL程序》

        在之前的博客SparkSQL系列中，已经大致为大家介绍了​​DataFrame​​​,​​DataSet​​​的概念以及它们之间与​​RDD​​之间的互转的操作描述。本篇博客，为大家带来的是关于如何在IDEA上创建SparkSQL程序，并实现数据查询与(DataFrame,DataSet,RDD)互相转换的功能!

文章目录

• ​​IDEA中创建SparkSQL程序​​

• ​​<1>数据查询​​
• ​​<2>类型转化​​

IDEA中创建SparkSQL程序

        IDEA中程序的打包和运行方式都和SparkCore类似。首先Maven依赖中需要添加新的依赖项：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.11</artifactId>
    <version>2.1.1</version>
</dependency>

准备数据源people.json

{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}

<1>数据查询

object SparkSQL01_Demo {

  def main(args: Array[String]): Unit = {

    // 创建配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL01_Demo")

    // 创建 SparkSQL的环境变量
    // SparkConf()
    val spark = SparkSession
      .builder()
      .config(sparkConf)
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._

    val df = spark.read.json("in/people.json")

    // 查询所有数据
    df.show()

    // 过滤器查询
    df.filter($"age">21).show()

    // 创建临时表
    df.createOrReplaceTempView("persons")

    // Sparksql 查询
    spark.sql("SELECT * FROM persons where age > 21").show()
    
    // 释放资源
    spark.stop()

  }
}

<2>类型转化

object SparkSQL02_SQL {

  def main(args: Array[String]): Unit = {

    // 创建配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL01_Demo")

    // 创建 SparkSQL的环境变量
    // SparkSession()
    val spark:SparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .getOrCreate()


    // 导入隐式转换
    // 这里的spark不是包名的含义,是SparkSession对象的名字
    import spark.implicits._


    // 创建RDD
  val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List((1,"zhangsan",20),(2,"lisi",30),(3,"wangwu",40)))


    // 转换为DF,需给出结构
    val df: DataFrame = rdd.toDF("id","name","age")

    // 转换为DS,需给出类型
    val ds: Dataset[User] = df.as[User]

    // 转换为DF
    val df1: DataFrame = ds.toDF()

    // 转换为RDD
    val rdd1: RDD[Row] = df1.rdd

    //访问
    rdd1.foreach(row=>{
      // 获取数据时,可以通过索引访问数据
      println(row.getInt(0))
    })


    // 释放资源
    spark.stop()

  }

  // 定义一个样例类
  case class User(id:Int,name:String,age:Int)
}

        本次的分享就到这里，受益或感兴趣的朋友记得点个赞支持一下~