Spark系列--SparkSQL(三)执行SparkSQL查询

叁歲伎倆 2022-05-19 08:15 514阅读 0赞

## 前言 ##

**SparkSession**

> 在老的版本中，SparkSQL提供两种SQL查询起始点，一个叫SQLContext，用于Spark自己提供的SQL查询，一个叫HiveContext，用于连接Hive的查询，SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext，所以计算实际上是由sparkContext完成的。

import org.apache.spark.sql.SparkSession  
    
    val spark = SparkSession .builder() .appName("Spark SQL basic example") .config("spark.some.config.option", "some-value") .getOrCreate()  
    
    // For implicit conversions like converting RDDs to DataFrames 
    import spark.implicits._

*  `SparkSession.builder` 用于创建一个SparkSession。
 *  `import spark.implicits._`的引入是用于将DataFrames隐式转换成RDD，使df能够使用RDD中的方法。

如果需要Hive支持，则需要以下创建语句(`.enableHiveSupport()` )：

import org.apache.spark.sql.SparkSession  
    
    val spark = SparkSession .builder() .appName("Spark SQL basic example") .config("spark.some.config.option", "some-value") .enableHiveSupport() .getOrCreate()  
    
    // For implicit conversions like converting RDDs to DataFrames
     import spark.implicits._

## 一、Spark-Shell查询 ##

**（1）启动Spark-Shell(本人是以Standlone模式启动Spark集群的）**  
**（2）创建测试JSON数据，注意每一行是一个JSON数据**

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

**（3）上传测试JSON数据到Hadoop集群**

$ hadoop fs -put /opt/data/people.json /input

**（4）执行SparkSQL查询**

scala> val df = spark.read.json("/input/people.json")
    df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]                
    
    scala> df.show +----+-------+
    | age| name| +----+-------+
    |null|Michael|
    |  30|   Andy|
    | 19| Justin| +----+-------+
    
    //注册临时表/视窗，进行sql语句操作
    scala> df.createOrReplaceTempView("people")
    
    scala> spark.sql("SELECT * FROM people").show() +----+-------+
    | age| name| +----+-------+
    |null|Michael|
    |  30|   Andy|
    | 19| Justin| +----+-------+

## 二、IDEA创建SparkSQL程序 ##

[Spark系列(五)IDEA编写及调试Spark的WordCount程序][Spark_IDEA_Spark_WordCount]

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

<dependency> 
        <groupId>org.apache.spark</groupId> 
        <artifactId>spark-sql_2.11</artifactId> 
        <version>${spark.version}</version> 
        
     </dependency>

如果连接HIVE还需要添加依赖:

<dependency> 
        <groupId>org.apache.spark</groupId> 
        <artifactId>spark-hive_2.11</artifactId> 
        <version>${spark.version}</version> 
        
     </dependency>

import org.apache.spark.SparkConf
    import org.apache.spark.sql.{DataFrame, SparkSession}
    
    object HelloWorld extends App { 
    
      private val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("helloworld")
    
      private val session: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    
      val sc = session.sparkContext
    
      import session.implicits._
    
      private val df: DataFrame = session.read.json("/home/parallels/Documents/m-spark/spark-sql/sparksql-helloworld/src/main/resources/people.json")
    
      df.show()
    
      session.close()
    
    }

直接右键RUN，运行结果如下：

+----+-------+
    | age| name| +----+-------+
    |null|Michael|
    |  30|   Andy|
    | 19| Justin| +----+-------+

## 三、运行Spark SQL CLI ##

Spark SQL CLI可以很方便的在本地运行Hive元数据服务以及从命令行执行查询任务。需要注意的是，Spark SQL CLI不能与Thrift JDBC服务交互。 在Spark目录下执行如下命令启动Spark SQL CLI：

./bin/spark-sql

**配置Hive需要替换 conf/ 下的 hive-site.xml 。**

[Spark_IDEA_Spark_WordCount]: https://blog.csdn.net/u012834750/article/details/81016433