spark-sql实践

Dear 丶 2022-03-02 16:10 270阅读 0赞

### 实践一：读取hdfs方式访问数据 ###

1.数据准备：

![20190320205446734.png][]

2.spark-sq代码：

package com.badou.sql
    
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.sql.Row
    import org.apache.spark.sql.types.{StringType, StructField, StructType}
    
    import scala.collection.mutable
    
    object sqlJsonText {
      def main(args: Array[String]) {
    
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("sqlJsonText")
        val sc = new SparkContext(sparkConf)
    
        val sqlContext = new org.apache.spark.sql.SQLContext(sc)
        val personInfo = sqlContext.read.json("hdfs://master:9000/person_info.json")
    
        personInfo.registerTempTable("personInfo")
    
        sqlContext.sql("SELECT id, name, age FROM personInfo").show()
    
        println(personInfo.schema)
      }
    }

idea运行代码结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70][]

实践2：读取hive数据：

spark-sql代码：

package com.badou.sql
    
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.sql.hive.HiveContext
    
    object sqlHiveTest {
      def main(args: Array[String]) {
    
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("sqlHiveTest")
        val sc = new SparkContext(sparkConf)
    
        val hiveContext = new HiveContext(sc)
        hiveContext.table("movie_table").registerTempTable("movie_table")
    
        hiveContext.sql("SELECT movieid, title FROM movie_table").show()
      }
    }

脚本：

![20190320211436742.png][]

执行结果：

![20190320211532211.png][]

实践3：sqlUdf

代码：

package com.badou.sql
    
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.sql.DataFrame
    import org.apache.spark.sql.Row
    import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
    import org.apache.spark.sql.expressions.MutableAggregationBuffer
    import org.apache.spark.sql.hive.HiveContext
    import org.apache.spark.sql.types._
    
    
    class  WordcountUdaf extends UserDefinedAggregateFunction {
      // 该方法指定具体输入数据的类型
      override def inputSchema: StructType = StructType(Array(StructField("input", StringType, true)))
      //在进行聚合操作的时候所要处理的数据的结果的类型
      override def bufferSchema: StructType = StructType(Array(StructField("count", IntegerType, true)))
      //指定UDAF函数计算后返回的结果类型
      override def dataType: DataType = IntegerType
      // 确保一致性 一般用true
      override def deterministic: Boolean = true
      //在Aggregate之前每组数据的初始化结果
      override def initialize(buffer: MutableAggregationBuffer): Unit = {buffer(0) =0}
      // 在进行聚合的时候，每当有新的值进来，对分组后的聚合如何进行计算
      // 本地的聚合操作，相当于Hadoop MapReduce模型中的Combiner
      override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
        buffer(0) = buffer.getAs[Int](0) + 1
      }
      //最后在分布式节点进行Local Reduce完成后需要进行全局级别的Merge操作
      override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
        buffer1(0) = buffer1.getAs[Int](0) + buffer2.getAs[Int](0)
      }
      //返回UDAF最后的计算结果
      override def evaluate(buffer: Row): Any = buffer.getAs[Int](0)
    }
    
    object sqlUdaf {
      def main(args: Array[String]) {
    
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("sqlJsonText")
        val sc = new SparkContext(sparkConf)
    
        val hiveContext = new HiveContext(sc)
        hiveContext.table("rating_table").registerTempTable("rating_table")
    
        hiveContext.udf.register("strlen", (input: String) => input.length)
        hiveContext.udf.register("wordCount", new WordcountUdaf)
    
        hiveContext.sql("select rating, wordCount(rating) as count, strlen(rating) as length" +
          " from rating_table group by rating").show()
      }
    }

结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 1][]

实践4：udaf（计算音乐打分的有多少人）

代码：

执行脚本：

![20190320213245650.png][]

结果：

![20190320213219175.png][]

实践5：进入sql

![20190320214000128.png][]

在hive和在sql中，发现在sql中执行效率远超过hive：

select rating, count(\*) from rating\_table\_ex group by rating limit 100;

[20190320205446734.png]: /images/20220302/60ba24da5e514835a6d752aafe096603.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70]: /images/20220302/2d8d9578e8f34d3993eab7a6cc20c065.png
[20190320211436742.png]: /images/20220302/257696a846db4aea8ef0052b415edcce.png
[20190320211532211.png]: /images/20220302/b84c9f5bb974432fb8095f5929212cb6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 1]: /images/20220302/9fd5de5f66d34917b3f8d3140d5f3adc.png
[20190320213245650.png]: /images/20220302/8843f7291c464b4fa756b5c79c0912f2.png
[20190320213219175.png]: /images/20220302/c26b86ea10d94f3dbaaa31b44d00645b.png
[20190320214000128.png]: /images/20220302/aff2cb8c8287444f8d793384e7e7c5e2.png