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

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

IDEA中创建SparkSQL程序

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

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

  

 

  
1
  
2
  
3
  
4
  
5
 

准备数据源people.json

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

  

 

  
1
  
2
  
3
 

<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()

  }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
  
30
  
31
  
32
  
33
  
34
  
35
  
36
  
37
 

<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)
}

  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
  
30
  
31
  
32
  
33
  
34
  
35
  
36
  
37
  
38
  
39
  
40
  
41
  
42
  
43
  
44
  
45
  
46
  
47
  
48
  
49
  
50
  
51
 

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

文章来源: alice.blog.csdn.net，作者：大数据梦想家，版权归原作者所有，如需转载，请联系作者。

原文链接：alice.blog.csdn.net/article/details/104711778