Spark SQL

天津储鑫盛钢材现货供应商  金牌会员 | 2024-7-13 02:29:04 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 809|帖子 809|积分 2427



Spark SQL

一、Spark SQL架构



  • 能够直接访问现存的Hive数据
  • 提供JDBC/ODBC接口供第三方工具借助Spark进行数据处置惩罚
  • 提供更高层级的接口方便处置惩罚数据
  • 支持多种操纵方式：SQL、API编程

    • API编程：Spark SQL基于SQL开辟了一套SQL语句的算子，名称和标准的SQL语句相似

  • 支持Parquet、CSV、JSON、RDBMS、Hive、HBase等多种外部数据源。(掌握多种数据读取方式)

  • Spark SQL核心：是RDD+Schema(算子+表结构)，为了更方便我们操纵，会将RDD+Schema发给DataFrame
  • 数据回灌：用于将处置惩罚和清洗后的数据回写到Hive中，以供后续分析和使用。
  • BI Tools：主要用于数据出现。
  • Spark Application：开辟人员使用Spark Application编写数据处置惩罚和分析逻辑，这些应用可以用不同的编程语言编写，比如Python、Scala、Java等。
二、Spark SQL运行原理




  • Catalyst优化器的运行流程：

  • Frontend（前端）

    • 输入：用户可以通过SQL查询或DataFrame API来输入数据处置惩罚逻辑。
    • Unresolved Logical Plan（未解析的逻辑筹划）：输入的SQL查询或DataFrame转换操纵会首先被转换为一个未解析的逻辑筹划，这个筹划包罗了用户哀求的所有操纵，但其中的表名和列名等可能尚未解析。

  • Catalyst Optimizer（Catalyst优化器） Catalyst优化器是Spark SQL的核心组件，它负责将逻辑筹划转换为物理执行筹划，并进行优化。Catalyst优化器包括以下几个阶段：

    • Analysis（分析）：将未解析的逻辑筹划中的表名和列名解析为详细的元数据，这一步依赖于Catalog（元数据存储）。输出是一个解析后的逻辑筹划。
    • Logical Optimization（逻辑优化）：对解析后的逻辑筹划进行各种优化，如投影剪切、过滤下推等。优化后的逻辑筹划更加高效。
    • Physical Planning（物理筹划）：将优化后的逻辑筹划转换为一个或多个物理执行筹划。每个物理筹划都代表了一种可能的执行方式。
    • Cost Model（本钱模型）：评估不同物理筹划的执行本钱，选择代价最低的物理筹划作为最终的物理筹划。

  • Backend（后端）

    • Code Generation（代码生成）：将选择的物理筹划转换为可以在Spark上执行的RDD操纵。这一步会生成现实的执行代码。
    • RDDs：最终生成的RDD操纵被执行，以完成用户哀求的数据处置惩罚使命。



  • 一个SQL查询在Spark SQL中的优化流程
  1. SELECT name FROM(
  2.   SELECT id, name FROM people
  3. ) p
  4. WHERE p.id = 1
复制代码



  • Filter下压：将Filter操纵推到更靠近数据源的位置，以减少不必要的数据处置惩罚。
  • 归并Projection：减少不必要的列选择
  • IndexLookup return:name：假如存在索引，可以直接通过索引查找并返回name列
三、Spark SQL API


  • SparkContext：Spark应用的主入口，代表了与Spark集群的连接。
  • SQLContext：Spark SQL的编程入口，使用SQLContext可以运行SQL查询、加载数据源和创建DataFrame。
  • HiveContext：SQLContext的一个子集，可以执行HiveQL查询，并且可以访问Hive元数据和UDF。
  • SparkSession：Spark2.0后推荐使用，归并了SQLContext和HiveContext，提供了与Spark所有功能交互的单一入口点。
    创建一个SparkSession就包罗了一个SparkContext。
  • 若同时需要创建SparkContext和SparkSession，必须先创建SparkContext再创建SparkSession。否则，会抛出如下异常，提示重复创建SparkContext：
详细解释

创建SparkSession的代码

  1. val conf: SparkConf = new SparkConf()
  2.         .setMaster("local[4]")
  3.         .setAppName("SparkSql")
  4. def main(args: Array[String]): Unit = {
  5.         SparkSession.builder()
  6.                 .config(conf)
  7.                 .getOrCreate()
  8. }
复制代码
优化：减少创建代码，SparkSessionBuilder工具类

  1. package com.ybg
  3. import org.apache.spark.{SparkConf, SparkContext}
  4. import org.apache.spark.sql.SparkSession
  6. // 封装SparkSession的创建方法
  7. class SparkSessionBuilder(master:String,appName:String){
  8.   lazy val config:SparkConf = {
  9.     new SparkConf()
  10.       .setMaster(master)
  11.       .setAppName(appName)
  12.   }
  14.   lazy val spark:SparkSession = {
  15.     SparkSession.builder()
  16.       .config(config)
  17.       .getOrCreate()
  18.   }
  20.   lazy val sc:SparkContext = {
  21.     spark.sparkContext
  22.   }
  24.   def stop(): Unit = {
  25.     if (null != spark) {
  26.       spark.stop()
  27.     }
  28.   }
  29. }
  30. object SparkSessionBuilder {
  31.   def apply(master: String, appName: String): SparkSessionBuilder = new SparkSessionBuilder(master, appName)
  32. }
复制代码
四、Spark SQL依赖

pom.xml

  1. <properties>
  2.   <maven.compiler.source>8</maven.compiler.source>
  3.   <maven.compiler.target>8</maven.compiler.target>
  4.   <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  5.   <spark.version>3.1.2</spark.version>
  6.   <spark.scala.version>2.12</spark.scala.version>
  7.   <hadoop.version>3.1.3</hadoop.version>
  8.   <mysql.version>8.0.33</mysql.version>
  9.   <hive.version>3.1.2</hive.version>
  10.   <hbase.version>2.3.5</hbase.version>
  11.   <jackson.version>2.10.0</jackson.version>
  12. </properties>
  14. <dependencies>
  15.   <!-- spark-core -->
  16.   <dependency>
  17.     <groupId>org.apache.spark</groupId>
  18.     <artifactId>spark-core_${spark.scala.version}</artifactId>
  19.     <version>${spark.version}</version>
  20.   </dependency>
  22.   <!-- spark-sql -->
  23.   <dependency>
  24.     <groupId>org.apache.spark</groupId>
  25.     <artifactId>spark-sql_${spark.scala.version}</artifactId>
  26.     <version>${spark.version}</version>
  27.   </dependency>
  28.   若出现如下异常：
  29.   Caused by: com.fasterxml.jackson.databind.JsonMappingException:
  30.   Scala module 2.10.0 requires Jackson Databind version >= 2.10.0 and < 2.11.0
  31.     追加如下依赖：
  32.         -->
  33.   <!-- jackson-databind -->
  34.   <dependency>
  35.       <groupId>com.fasterxml.jackson.core</groupId>
  36.       <artifactId>jackson-databind</artifactId>
  37.       <version>2.10.0</version>
  38.   </dependency>
  39.   
  40.   <!-- mysql -->
  41.   <dependency>
  42.     <groupId>com.mysql</groupId>
  43.     <artifactId>mysql-connector-j</artifactId>
  44.     <version>${mysql.version}</version>
  45.   </dependency>
  46. </dependencies>
复制代码
log4j.properties

log4j.properties应该放在资源包下。
  1. log4j.rootLogger=ERROR, stdout, logfile # 设置可显示的信息等级
  2. log4j.appender.stdout=org.apache.log4j.ConsoleAppender
  3. log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
  4. log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
  5. log4j.appender.logfile=org.apache.log4j.FileAppender
  6. log4j.appender.logfile.File=log/spark_first.log
  7. log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
  8. log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
复制代码
五、Spark SQL数据集

1、DataSet



  • 简介：

    • 从Spark 1.6开始引入的新的抽象。
    • 是特定领域对象中的强类型集合。
    • 可以使用函数式编程或SQL查询进行操纵。
    • 即是RDD + Schema。

2、DataFrame



  • 简介：

    • DataFrame是特殊的DataSet：DataFrame=DataSet[Row]，行对象的集合，每一行就是一个行对象。
    • 类似于传统数据的二维表格。

  • 特性：

    • Schema：在RDD基础上增加了Schema，描述数据结构信息
    • 嵌套数据类型：支持struct,map,array等嵌套数据类型。
    • API：提供类似SQL的操纵接口。

详细解释

创建DataSet的代码

  1. val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
  2. // 提供了一组隐式转换，这些转换允许将Scala的本地集合类型(如Seq、Array、List等)转换为Spark的DataSet。
  3. import spark.implicits._
  4. val dsPhone: Dataset[Product] = spark.createDataset(Seq(
  5.   Product(1, "Huawei Mate60", 5888.0f),
  6.   Product(2, "IPhone", 5666.0f),
  7.   Product(3, "OPPO", 1888.0f)
  8. ))
  9. dsPhone.printSchema()
  10. /**
  11. * root
  12. * |-- id: integer (nullable = false)
  13. * |-- name: string (nullable = true)
  14. * |-- price: float (nullable = false)
  15. */
复制代码


创建DataFrame的代码



  • 读取CSV文件

    • 对于CSV文件，在构建DataFrame之前，必须要先创建一个Schema，再根据文件类型分不怜悯况进行导入。(读取JSON文件或者数据库表都并不需要)
    • 注意：必须要import spark.implicits._，导入隐式类，才能够识别一些隐式转换，否则会报错。
    • CSV文件在创建DataFrame时，可以选择尽量模仿Hive中的OpenCSVSerDe的

  1. val spark: SparkSession = SparkSession.builder()
  2. .config(conf)
  3. .getOrCreate()
  4. import spark.implicits._
  5. val schema: StructType = StructType(
  6.   Seq(
  7.     StructField("user_id", LongType),
  8.     StructField("locale", StringType),
  9.     StructField("birthYear", IntegerType),
  10.     StructField("gender", StringType),
  11.     StructField("joinedAt", StringType),
  12.     StructField("location", StringType),
  13.     StructField("timezone", StringType)
  14.   )
  15. )
  16. val frmUsers: DataFrame = spark.read
  17. .schema(schema)
  18. .option("separator", ",") // 指定文件分割符
  19. .option("header", "true") // 指定CSV文件包含表头
  20. .option("quoteChar", """)
  21. .option("escapeChar", "\")
  22. .csv("C:\\Users\\lenovo\\Desktop\\users.csv")
  23. .repartition(4)
  24. .cache()
复制代码


  • 读取JSON文件
  1. val frmUsers2: DataFrame = spark.read.json("hdfs://single01:9000/spark/cha02/users.json")
  2. frmUsers2.show()
复制代码


  • 读取数据库表
  1. val url = "jdbc:mysql://single01:3306/test_db_for_bigdata" // 数据库连接地址
  2. val mysql = new Properties()
  3. mysql.setProperty("driver", "com.mysql.cj.jdbc.Driver")
  4. mysql.setProperty("user", "root")
  5. mysql.setProperty("password", "123456")
  6. spark
  7.   .read
  8.   .jdbc(url,"test_table1_for_hbase_import",mysql) // (url,TableName,连接属性)
  9.   .show(100)
复制代码
六、Spark_SQL的两种编码方式

  1. val spark: SparkSession = SparkSession.builder()
  2. .config(conf)
  3. .getOrCreate()
  4. import spark.implicits._
  5. val schema: StructType = StructType(
  6.   Seq(
  7.     StructField("user_id", LongType),
  8.     StructField("locale", StringType),
  9.     StructField("birthYear", IntegerType),
  10.     StructField("gender", StringType),
  11.     StructField("joinedAt", StringType),
  12.     StructField("location", StringType),
  13.     StructField("timezone", StringType)
  14.   )
  15. )
  16. val frmUsers: DataFrame = spark.read
  17. .schema(schema)
  18. .option("separator", ",") // 指定文件分割符
  19. .option("header", "true") // 指定CSV文件包含表头
  20. .option("quoteChar", """)
  21. .option("escapeChar", "\")
  22. .csv("C:\\Users\\lenovo\\Desktop\\users.csv")
  23. .repartition(4)
  24. .cache()
复制代码
此处已经创建好了DataFrame
1. 面向标准SQL语句（偷懒用）

  1. frmUsers.registerTempTable("user_info") // 此方法已过期
  2. spark.sql(
  3.       """
  4.         |select * from user_info
  5.         |where gender='female'
  6.         |""".stripMargin)
  7.                 .show(10)
复制代码
2. 使用Spark中的SQL算子（更规范）

  1. frmUsers
  2.       .where($"birthYear">1990)
  3.       .groupBy($"locale")
  4.       .agg(
  5.         count($"locale").as("locale_count"),
  6.         round(avg($"birthYear"),2).as("avg_birth_year")
  7.       )
  8.       .where($"locale_count">=10 and $"avg_birth_year">=1993)
  9.       .orderBy($"locale_count".desc)
  10.       .select(
  11.         $"locale", $"locale_count", $"avg_birth_year",
  12.         dense_rank()
  13.           .over(win)
  14.           .as("rnk_by_locale_count"),
  15.         lag($"locale_count",1)
  16.           .over(win)
  17.           .as("last_locale_count")
  18.       )
  19.       .show(10)
复制代码
七、常用算子

1.基本SQL模板

  1. select
  2.                 col,cols*,agg*
  3. where
  4.                 conditionCols
  5. group by
  6.                 col,cols*
  7. having
  8.                 condition
  9. order by
  10.                 col asc|desc
  11. limit
  12.                 n
复制代码
2.select

select语句在代码的开头可以不写，因为有后续的类似where和group by语句已经对列进行了操纵，指明确列名。假如后续有select语句，则优先按照后面的select语句进行。
  1. frmUsers.select(
  2.         $"locale",$"locale_count"
  3. )
复制代码
3.agg

  1. .agg(
  2.   count($"locale").as("locale_count"),
  3.   round(avg($"birthYear"),2).as("avg_birth_year")
  4. )
复制代码
4.窗口函数



  • over子句
注意：over子句中的分区信息是可以被重用的
  1. val win: WindowSpec = Window.partitionBy($"gender").orderBy($"locale_count".desc)
  2. frmUsers
  3.   ...
  4.   .select(
  5.                 dense_rank()
  6.                   .over(win)
  7.                   .as("rnk_by_locale_count")
  8.         )
复制代码
5.show

show(N)表示显示符合条件的至多N条数据。(不是取前N条再提取出其中符合条件的数据)
  1. frmUsers
  2.   ...
  3.   .show(10)
复制代码
6.条件筛选 where

  1. newCol:Column = $"cus_state".isNull
  2. newCol:Column = $"cus_state".isNaN
  3. newCol:Column = $"cus_state".isNotNull
  5. newCol:Column = $"cus_state".gt(10)                <=>        $"cus_state">10
  6. newCol:Column = $"cus_state".geq(10)        <=>        $"cus_state">=10
  7. newCol:Column = $"cus_state".lt(10)                <=>        $"cus_state"<10
  8. newCol:Column = $"cus_state".leq(10)        <=>        $"cus_state"<=10
  9. newCol:Column = $"cus_state".eq(10)                <=>        $"cus_state"===10
  10. newCol:Column = $"cus_state".ne(10)                <=>        $"cus_state"=!=10
  11. newCol:Column = $"cus_state".between(10,20)
  13. newCol:Column = $"cus_state".like("张%")
  14. newCol:Column = $"cus_state".rlike("\\d+")
  16. newCol:Column = $"cus_state".isin(list:Any*)
  17. newCol:Column = $"cus_state".isInCollection(values:Itrable[_])
  19. 多条件：
  20. newCol:Column = ColOne and ColTwo
  21. newCol:Column = ColOne or ColTwo
复制代码
在Spark SQL中，不存在Having子句，Where子句的现实作用根据相对于分组语句的前后决定。
7.分组

  1. // 多重分组
  2. /**
  3. rollup的效果：
  4. select birthYear,count(*) from user group by birthYear
  5. union all
  6. select gender,birthYear,count(*) from user group by gender,birthYear
  7. 存在"字段不对应"的情况：
  8. 空缺的字段会自动补全为null
  9. */
  10. frmUsers
  11.         .rollup("gender", "birthYear")
  12.         .count()
  13.         .show(100)
复制代码
  1. // 为了方便查找到每个数据行所对应的分组方式
  2. spark.sql(
  3.   """
  4.   |select grouping__id,gender,birthYear,count(8) as cnt from user_info
  5.   |group by gender,birthday,
  6.   |grouping sets(gender,birthday,(gender,birthYear))
  7.   |""".stripMargin)
  8. .show(100)
  9. // 这里的group by子句定义了分组的列，到grouping sets明确指定了分组的组合
  10. // 因而，在数仓设计的过程中，我们能够对不同分组依据下的不同数据依据grouping__id做分区。
复制代码


  • RollUp和Cube的区别
    假设有三列：1, 2, 3，使用CUBE(1, 2, 3)，会生成以下组合：

    • GROUP BY ()（不分组，整体聚合）
    • GROUP BY (1)
    • GROUP BY (2)
    • GROUP BY (3)
    • GROUP BY (1, 2)
    • GROUP BY (1, 3)
    • GROUP BY (2, 3)
    • GROUP BY (1, 2, 3)
    ROLLUP生成的分组组合是层级的，它从最详细的分组开始，一步步减少分组的列，直到整体聚合。
    假设有三列：1, 2, 3，使用ROLLUP(1, 2, 3)，会生成以下组合：
       

    • GROUP BY (1, 2, 3)（最详细的分组）
    • GROUP BY (1, 2)
    • GROUP BY (1)
    • GROUP BY ()（不分组，整体聚合）

8.关联查询

  1. val frmClass: DataFrame = spark.createDataFrame(
  2.   Seq(
  3.     Class(1, "yb12211"),
  4.     Class(2, "yb12309"),
  5.     Class(3, "yb12401")
  6.   )
  7. )
  9. val frmStu: DataFrame = spark.createDataFrame(
  10.   Seq(
  11.     Student("henry", 1),
  12.     Student("ariel", 2),
  13.     Student("jack", 1),
  14.     Student("rose", 4),
  15.     Student("jerry", 2),
  16.     Student("mary", 1)
  17.   )
  18. )
  20. // 1.笛卡尔积(默认情况下)
  21. frmStu.as("S")
  22. .join(frmClass.as("C"))
  23. .show(100)
  24. /**
  25. +-----+-------+-------+---------+
  26. | name|classId|classId|className|
  27. +-----+-------+-------+---------+
  28. |henry|    1 |    1 |  yb12211|
  29. |henry|    1 |    2 |  yb12309|
  30. |henry|    1 |    3 |  yb12401|
  31. |ariel|    2 |    1 |  yb12211|
  32. |ariel|    2 |    2 |  yb12309|
  33. |ariel|    2 |    3 |  yb12401|
  34. | jack|    1 |    1 |  yb12211|
  35. | jack|    1 |    2 |  yb12309|
  36. | jack|    1 |    3 |  yb12401|
  37. | rose|    4 |    1 |  yb12211|
  38. | rose|    4 |    2 |  yb12309|
  39. | rose|    4 |    3 |  yb12401|
  40. |jerry|    2 |    1 |  yb12211|
  41. |jerry|    2 |    2 |  yb12309|
  42. |jerry|    2 |    3 |  yb12401|
  43. | mary|    1 |    1 |  yb12211|
  44. | mary|    1 |    2 |  yb12309|
  45. | mary|    1 |    3 |  yb12401|
  46. +-----+-------+-------+---------+
  47. */
  48. // 2.内连接
  49. frmStu.as("S")
  50. .join(frmClass.as("C"), $"S.classId" === $"C.classId","inner")
  51. .show(100)
  52. /**
  53. +-----+-------+-------+---------+
  54. | name|classId|classId|className|
  55. +-----+-------+-------+---------+
  56. |henry|    1 |    1 |  yb12211|
  57. |ariel|    2 |    2 |  yb12309|
  58. | jack|    1 |    1 |  yb12211|
  59. |jerry|    2 |    2 |  yb12309|
  60. | mary|    1 |    1 |  yb12211|
  61. +-----+-------+-------+---------+
  62. */
  63. // 启用using：使用Seq("Column")代表关联字段
  64. frmStu.as("S")
  65. .join(frmClass.as("C"), Seq("classId"),"right")
  66. .show(100)
  68. // 3.外连接
  69. frmStu.as("S")
  70. .join(frmClass.as("C"), $"S.classId" === $"C.classId","outer") // left | right | outer
  71. .show(100)
  72. /**
  73. +-----+-------+-------+---------+
  74. | name|classId|classId|className|
  75. +-----+-------+-------+---------+
  76. |henry|    1 |    1 |  yb12211|
  77. | jack|    1 |    1 |  yb12211|
  78. | mary|    1 |    1 |  yb12211|
  79. | null|  null |    3 |  yb12401|
  80. | rose|    4 |  null |    null|
  81. |ariel|    2 |    2 |  yb12309|
  82. |jerry|    2 |    2 |  yb12309|
  83. +-----+-------+-------+---------+
  84. */
  85. // 4.反连接：返回左数据集中所有没有关联字段匹配记录的左数据集的行
  86. frmStu.as("S")
  87. .join(frmClass.as("C"), $"S.classId" === $"C.classId","anti")
  88. .show(100)
  89. /**
  90. +----+-------+
  91. |name|classId|
  92. +----+-------+
  93. |rose|    4 |
  94. +----+-------+
  95. */
  96. // 5.半连接：返回左数据集中所有有关联字段匹配记录的左数据集的行
  97. frmStu.as("S")
  98. .join(frmClass.as("C"), $"S.classId" === $"C.classId","semi")
  99. .show(100)
  100. /**
  102. +-----+-------+
  103. | name|classId|
  104. +-----+-------+
  105. |henry|    1 |
  106. |ariel|    2 |
  107. | jack|    1 |
  108. |jerry|    2 |
  109. | mary|    1 |
  110. +-----+-------+
  111. */
复制代码
9.排序

  1. frmStu.orderBy(cols:Column*)
复制代码
10.数据截取

  1. frmStu.tail(n:Int)
  2. frmStu.take(n:Int)
复制代码


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

天津储鑫盛钢材现货供应商

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表