spark sql 快速体验调试

雨点打透心脏的1/2处 2022-04-06 02:30 229阅读 0赞

spark sql提供了更快的查询性能，如何能够更快的体验，开发和调试spark sql呢？按照正规的步骤我们一般会集成hive，然后使用hive的元数据查询hive表进行操作，这样以来我们还需要考虑跟hive相关的东西，如果我们仅仅是学习spark sql查询功能，那么仅仅使用IDEA的IDE环境即可，而且能够在win上快速体验，不需要hive数据仓库，我们直接使用数组造点数据，然后转成DF，最后直接使用spark sql操作即可。  
  
首先，看下pom文件的核心依赖：

<dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql_2.11</artifactId>
                <version>2.1.0</version>
            </dependency>
    
    
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-hive_2.11</artifactId>
                <version>2.1.0</version>
            </dependency>
    
    
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-library</artifactId>
                <version>2.11.8</version>
                <scope>compile</scope>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>2.7.3</version>
            </dependency>
    
    
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.11</artifactId>
                <version>2.1.0</version>
            </dependency>

然后看一个例子spark sql的测试例子：

import org.apache.spark.sql.SparkSession
    /**
      * spark sql本地测试例子
      */
    object TestGroup {
       
    
      def main(args: Array[String]): Unit = {
       
    
        val spark = SparkSession
          .builder().master("local[1]")//设置loca模式
          .appName("Spark SQL basic example")//设置app的名字
          .getOrCreate()
    
    
        import spark.implicits._//导入隐式的转化函数
        import spark.sql //导入sql函数
    
        //使用Seq造数据，三列数据
        val df = spark.sparkContext.parallelize(Seq((0,"p",30.9),
          (0,"u",22.1),
          (1,"r",19.6),
          (2,"cat40",20.7),
          (2,"cat187",27.9),
          (2,"cat183",11.3),
          (3,"cat8",35.6))).toDF("id", "name", "price")//转化df的三列数据s
    
          df.createTempView("pro")//创建表明为pro
    
          //按照id分组，统计每组数量，统计每组里面最小的价格，然后收集每组里面的数据
          val ds=sql("select id, count(*) as c,min(price) as min_price,
    collect_list(struct(name, price)) as res  from pro   group by id ");
    
        ds.cache() //需要多次查询的数据，可以缓存起来
    
        //获取查询的结果，遍历获取结果集
        ds.select("id","c","res","min_price").collect().foreach(line=>{
       
    
          import org.apache.spark.sql.Row//导入Row对象
          val id=line.getAs[Int]("id")//获取id
          val count=line.getAs[Long]("c")//获取数量
          val min_price=line.getAs[Double]("min_price")//获取最小的价格
          val value=line.getAs[Seq[Row]]("res")//获取每组内的数据集合，注意是一个Row实体
          println(id+"  "+count+"  "+"  "+min_price)//打印数据
          value.foreach(row=>{//遍历组内数据集合，然后打印
            println(row.getAs[String]("name")+" "+row.getAs[Double]("price"))
    
          })
    
    
        })
    
        spark.stop()
    
      }
    
    }

至此，一个涵盖spark sql比较全的功能例子的小工程就完成了，上面的代码直接可在win上运行，而且里面的数据随时自己添加删除，以便于可以测试spark sql与预期效果对比，上面的sql中还用到了分组里面的高级用法，分组后，收集组内数据，注意组内数据收集，如果是单个字段，直接用collect\_list或者collect\_set即可，但是如果是多个字段，这个时候必须用到struct类型了，最终转化后的类型就是row的集合，里面的每个结构体会被转成一个row对象，一个组的数据，就是List<Row>了，最终可以在代码里面遍历取出。spark sql结合scala编程语言之后可以变得非常灵活，sql不擅长的就用编程语言解决的，sql擅长的就用sql方便快速得到数据，用起来非常干净清爽！  
  
  
\[b\]\[color=green\]\[size=large\]  
有什么问题可以扫码关注微信公众号：我是攻城师(woshigcs)，在后台留言咨询。  
技术债不能欠，健康债更不能欠， 求道之路，与君同行。  
\[/size\]\[/color\]\[/b\]  
\[img\]http://dl2.iteye.com/upload/attachment/0104/9948/3214000f-5633-3c17-a3d7-83ebda9aebff.jpg\[/img\]