Spark学习（4）-Spark SQL（2）

拼搏现实的明天。 2022-03-14 13:34 214阅读 0赞

**目录：**

1.  Spark SQL 使用
    
    1.  反射推断模式
    2.  用户自定义模式
    3.  两者对比
2.  Spark SQL 数据源

### 一.反射推断模式 ###

**1.使用反射推断模式的原因**

*  上一章在读取一个json的时候，对json文件的内容有没有要求？  
    文件内容必须是一个json格式的，就需要用到"字段名称":"值"，  
    spark sql将字段名称作为了dataFrame的schema。  
    如果我们要操作的目录是一个普通的文件，那么spark sql应该如何处理？  
    如果要实现从普通文件读取数据到spark sql当中，那么缺少的就是dataFrame的schema。  
    那么spark sql提供了一种解决方案：反射推断模式：借助样例类的方式来实现schema的定义。

**2.反射推断模式使用**

**核心代码**

//1.加入隐式转换    
        import sqlContext.implicits._
        //2.将普通文件转换RDD
        val UserRDD:RDD[User]=sc.textFile("e://user.txt").map(_.split(" ")).map(x=>User(x(0),x(1).toInt))
        //3.将RDD转换为DataFrame
        val df:DataFrame=UserRDD.toDF
        //4.创建虚表
        df.registerTempTable("user")
        //5.执行sql
        val sqlFrame=sqlContext.sql("select * from user")
        //6.结果展示
        sqlFrame.show()

**使用反射推断的时候需要注意以下方式：**

1.  将本地的一个文件转化为RDD（拿到文件-->拆分-->组合成样例类的格式）
2.  将转换之后的RDD又转换为DataFrame（.toDF）
3.  为DF注册一个临时表名
4.  调用sqlContext.sql（“sql语句”）就可以调用show方法了
5.  关闭sc

**3.示例：步骤**

1.  创建样例类  
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70][]
2.  创建文本文档user.txt
    
        aa 23
        bb 32
        cc 23
3.  编写代码
    
        package test
        import org.apache.spark.SparkConf
        import org.apache.spark.SparkContext
        import org.apache.spark.SparkContext._
        import org.apache.spark.rdd.RDD
        import org.apache.spark.sql.SQLContext
        import org.apache.spark.sql.DataFrame
        object SparkFirst {
          def main(args: Array[String]): Unit = {
            //1.创建一个SparkConf对象
            val conf=new SparkConf().setAppName("HelloSpark").setMaster("local[*]")
            //2.创建一个SparkConf对象
            val sc=new SparkContext(conf)
            //3.创建一个SQLContext对象
            val sqlContext=new SQLContext(sc)
            
            //4.引入包，import的效果从开始延伸到语句块的结束
            /*如果手里有一个RDD，需要通过spark sql的方式进行操作的话，那么就需要用到spark sql的函数。
        		但是这些函数，默认情况下RDD是没有的。所以在scala中有一个很特殊的功能——隐式转换
        		将本地文件内容转换成一个DataFrame*/
            import sqlContext.implicits._
            
        
            //5.将普通文件转换RDD
            val UserRDD:RDD[User]=sc.textFile("e://user.txt").map(_.split(" ")).map(x=>User(x(0),x(1).toInt))
            //6.将RDD转换为DataFrame
            val df:DataFrame=UserRDD.toDF
            //7.操作
            //创建虚表
            df.registerTempTable("user")
            val sqlFrame=sqlContext.sql("select * from user")
            sqlFrame.show()
            //8.关闭SparkContext
            sc.stop()
          }
        }
        
        case class User(name:String,age:Int){
          
        }
4.  运行结果
    
        +----+---+
        |name|age|
        +----+---+
        |  aa| 23|
        |  bb| 32|
        |  cc| 23|
        +----+---+

### 二.用户自定义模式 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 1][]

**简要步骤（查询操作）：**

1.  定义字段名
2.  构造StructType类型
3.  构造RDD\[Row\]
4.  创建DataFrame：让StructTyp和RDD\[Row\]关联
5.  注册表名
6.  执行sql
7.  关闭

**代码：**

package test
    import org.apache.spark.SparkConf
    import org.apache.spark.SparkContext
    import org.apache.spark.SparkContext._
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.SQLContext
    import org.apache.spark.sql.DataFrame
    import org.apache.spark.sql.Row
    
    object SparkFirst {
      def main(args: Array[String]): Unit = {
        
        val conf=new SparkConf().setAppName("HelloSpark").setMaster("local[*]")
        val sc=new SparkContext(conf)
        val sqlContext=new SQLContext(sc)
        val UserRDD:RDD[String]=sc.textFile("e://user.txt")
        import sqlContext.implicits._
        
        //1.引入包
        import org.apache.spark.sql.types.StringType
        import org.apache.spark.sql.types.StructField
        import org.apache.spark.sql.types.StructType
        //2.定义schema字段名
        val schemaString:String="name age"
        //3.构造StructType类型
        val structType:StructType=StructType(schemaString.split(" ").map(field=>StructField(field,StringType,true)))
        //4.将RDD转换为ROW
        val rowRDD:RDD[Row]=UserRDD.map(_.split("\t")).map(x=>Row(x(0),x(1)))
        //5.将schema与数据关联
        val df:DataFrame=sqlContext.createDataFrame(rowRDD,structType)
        //6.注册为虚表
        df.registerTempTable("user")
        //7.执行查询
        val sqlFrame=sqlContext.sql("select * from user")
        //8.展示结果
        sqlFrame.show()
        //9.关闭SparkContext
        sc.stop()
      }
    }

## ##

### 三.两者对比 ###

**反射推断和用户自定义的对比**

1.  在构造的时候不同：
    
    1.  反射推断的方式是采用case class的方式
    2.  用户自定义的方式是采用 structType 的方式
2.  主要解决的问题是：将一些没有schame信息的源文件，可以采用以上两个方式进行sql操作
3.  注意事项
    
    1.  用户自定义方式：字段名是写在一个string类型上的，如果需要加上某一个字段，只需要在字段串上加上字段就可以了。
    2.  反射推断的方式：字段名是需要在case class构造器上加上相应的字段。  
        但是要特别注意：该样例类最好只能被当前的对象操作。不要让其它的程序共用一个样例类。

## ##

### 四.spark SQL 数据源 ###

获得数据，存储数据的种类

[https://www.w3cschool.cn/spark\_sql/spark\_sql\_data\_sources.html][https_www.w3cschool.cn_spark_sql_spark_sql_data_sources.html]

[https://www.cnblogs.com/ywjy/p/7747482.html][https_www.cnblogs.com_ywjy_p_7747482.html]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70]: /images/20220313/b7070cb08b4c46b29d70f4ce348f8ec6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 1]: /images/20220313/3fce4b02709b40b4b482403397ed2b71.png
[https_www.w3cschool.cn_spark_sql_spark_sql_data_sources.html]: https://www.w3cschool.cn/spark_sql/spark_sql_data_sources.html
[https_www.cnblogs.com_ywjy_p_7747482.html]: https://www.cnblogs.com/ywjy/p/7747482.html