通过IDEA编写Spark SQL, 以编程方式执行Spark SQL查询, 使用Scala语言操作Spark SQL 25

冷不防 2023-06-04 13:58 42阅读 0赞

# 1. 使用Scala语言操作Spark SQL, 将RDD转为DataFrame #

*  前面我们学习了如何在Spark Shell中使用SQL完成查询，现在我们通过**IDEA编写Spark SQL查询程序**。

Spark官网提供了两种方法来实现从RDD转换得到DataFrame，

*  **第一种方法是利用反射机制**，推导包含某种类型的RDD，通过反射将其转换为指定类型的DataFrame，**适用于提前知道RDD的schema**。
 *  **第二种方法通过编程接口与RDD进行交互获取schema**，并动态创建DataFrame，在运行时决定列及其类型。

首先在maven项目的pom.xml中添加Spark SQL的依赖。

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

## 1.1 通过反射推断Schema ##

*  scala支持使用**case class类型导入RDD**转换为DataFrame，通过**case class创建schema**，case class的**参数名称**会被利用反射机制作为列名。
 *  这种RDD可以高效的转换为DataFrame并注册为表。
 *  代码如下, 如下代码包括DSL语言风格和SQL语言风格,两个风格操作SparkSQL

package com.acece.SparkSql
    import org.apache.spark.SparkContext
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{ DataFrame, SparkSession}
    
    /** * RDD转化成DataFrame:利用反射机制 */
    //todo:定义一个样例类Person
    case class Person(id:Int,name:String,age:Int) 
    
    object CaseClassSchema { 
    
      def main(args: Array[String]): Unit = { 
          //todo：1、构建sparkSession 指定appName和master的地址
        val spark: SparkSession = SparkSession.builder()
                                  .appName("CaseClassSchema")
                                  .master("local[2]").getOrCreate()
          //todo:2、从sparkSession获取sparkContext对象
          val sc: SparkContext = spark.sparkContext
          sc.setLogLevel("WARN")//设置日志输出级别
          //todo:3、加载数据
          val dataRDD: RDD[String] = sc.textFile("D:\\person.txt")
          //todo:4、切分每一行记录
          val lineArrayRDD: RDD[Array[String]] = dataRDD.map(_.split(" "))
          //todo:5、将RDD与Person类关联
          val personRDD: RDD[Person] = lineArrayRDD.map(x=>Person(x(0).toInt,x(1),x(2).toInt))
          //todo:6、创建dataFrame,需要导入隐式转换
          import spark.implicits._
          val personDF: DataFrame = personRDD.toDF()
    
        //todo-------------------DSL语法操作 start--------------
        //1、显示DataFrame的数据，默认显示20行
        personDF.show()
        //2、显示DataFrame的schema信息
        personDF.printSchema()
        //3、显示DataFrame记录数
        println(personDF.count())
        //4、显示DataFrame的所有字段
        personDF.columns.foreach(println)
        //5、取出DataFrame的第一行记录
        println(personDF.head())
        //6、显示DataFrame中name字段的所有值
        personDF.select("name").show()
        //7、过滤出DataFrame中年龄大于30的记录
        personDF.filter($"age" > 30).show()
        //8、统计DataFrame中年龄大于30的人数
        println(personDF.filter($"age">30).count())
        //9、统计DataFrame中按照年龄进行分组，求每个组的人数
        personDF.groupBy("age").count().show()
        //todo-------------------DSL语法操作 end-------------
    
        //todo--------------------SQL操作风格 start-----------
        //todo:将DataFrame注册成表
        personDF.createOrReplaceTempView("t_person")
        //todo:传入sql语句，进行操作
    
        spark.sql("select * from t_person").show()
    
        spark.sql("select * from t_person where name='zhangsan'").show()
    
        spark.sql("select * from t_person order by age desc").show()
        //todo--------------------SQL操作风格 end-------------
    
    
        sc.stop()
    	spark.stop()
      }
    }

## 1.2 通过StructType直接指定Schema ##

**当case class不能提前定义好时,可以通过以下三步创建DataFrame**

*  将RDD转换为包含row的RDD.
 *  基于**StructType类型创建schema**，与第一步创建的RDD相匹配.
 *  通过**SparkSession的createDataFrame**方法,对第一步的RDD应用schema创建DataFrame  
    **代码如下:**

package com.acece.SparkSql
    
    import org.apache.spark.SparkContext
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.types.{ IntegerType, StringType, StructField, StructType}
    import org.apache.spark.sql.{ DataFrame, Row, SparkSession}
    
    /** * RDD转换成DataFrame:通过指定schema构建DataFrame */
    object SparkSqlSchema { 
      def main(args: Array[String]): Unit = { 
          //todo:1、创建SparkSession,指定appName和master
          val spark: SparkSession = SparkSession.builder()
                                    .appName("SparkSqlSchema")
                                    .master("local[2]")
                                    .getOrCreate()
          //todo:2、获取sparkContext对象
        val sc: SparkContext = spark.sparkContext
          //todo:3、加载数据
        val dataRDD: RDD[String] = sc.textFile("D:\\person.txt")
          //todo:4、切分每一行
        val dataArrayRDD: RDD[Array[String]] = dataRDD.map(_.split(" "))
          //todo:5、加载数据到Row对象中
        val personRDD: RDD[Row] =  	 dataArrayRDD.map(x=>Row(x(0).toInt,x(1),x(2).toInt))
          //todo:6、创建schema
        val schema:StructType= StructType(Seq(
                                          StructField("id", IntegerType, false),
                                          StructField("name", StringType, false),
                                          StructField("age", IntegerType, false)
                                        ))
    
         //todo:7、利用personRDD与schema创建DataFrame
        val personDF: DataFrame = spark.createDataFrame(personRDD,schema)
    
        //todo:8、DSL操作显示DataFrame的数据结果
        personDF.show()
    
        //todo:9、将DataFrame注册成表
        personDF.createOrReplaceTempView("t_person")
        
        //todo:10、sql语句操作
        spark.sql("select * from t_person").show()
    
        spark.sql("select count(*) from t_person").show()
    
    
        sc.stop()
    	spark.stop()
      }
    }

# 2. 编写Spark SQL程序操作HiveContext #

*  **HiveContext**是对应spark-hive这个项目,与hive有部分耦合, 支持hql,是SqlContext的子类.
 *  在Spark2.0之后，HiveContext和SqlContext在**SparkSession进行了统一**，可以通过操作S**parkSession来操作HiveContext和SqlContext**。

## 2.1 添加pom依赖 ##

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-hive_2.11</artifactId>
        <version>2.0.2</version>
    </dependency>

## 2.2 代码实现 ##

package com.acece.SparkSql
    
    import org.apache.spark.sql.SparkSession
    /** * todo:Sparksql操作hive的sql */
    object HiveSupport { 
      def main(args: Array[String]): Unit = { 
          //todo:1、创建sparkSession
         val spark: SparkSession = SparkSession.builder()
           .appName("HiveSupport")
           .master("local[2]")
           .config("spark.sql.warehouse.dir", "d:\\spark-warehouse")
           .enableHiveSupport() //开启支持hive
           .getOrCreate()
        spark.sparkContext.setLogLevel("WARN")  //设置日志输出级别
    
    
        //todo:2、操作sql语句
    
        spark.sql("CREATE TABLE IF NOT EXISTS person (id int, name string, age int) row format delimited fields terminated by ' '")
        spark.sql("LOAD DATA LOCAL INPATH './data/student.txt' INTO TABLE person")
        spark.sql("select * from person ").show()
        spark.stop()
      }
    }

**需要在当前项目下创建一个data目录，然后在data目录下创建一个student.txt数据文件。**