SparkSQL 之 Shuffle Join 内核原理及应用深度剖析-Spark商业源码实战

时间  2019-11-24
标签 sparksql shuffle join 内核 原理 应用 深度 剖析 spark 商业 源码 实战 栏目 SQL 繁體版
原文   原文链接

本套技术专栏是做者（秦凯新）平时工做的总结和升华，经过从真实商业环境抽取案例进行总结和分享，并给出商业应用的调优建议和集群环境容量规划等内容，请持续关注本套博客。版权声明：禁止转载，欢迎学习。QQ邮箱地址：1120746959@qq.com，若有任何商业交流，可随时联系。算法

1 Spark SQL 坚实后盾DataFrame

  • DataFrame是一个分布式数据容器，更像传统数据库的二维表格，除了数据之外，还掌握数据的结构信息，即schema。同时，与Hive相似，DataFrame也支持嵌套数据类型（struct、array和map）。
  • JSON schema自动推导
  • Hive风格分区表自动识别
  • 充分利用RCFile、ORC、Parquet等列式存储格式的优点，仅扫描查询真正涉及的列，忽略其他列的数据。
  • 聚合统计函数支持

2 Spark SQL 源码包结构(溯本逐源)

主要分为4类：sql

  • core模块：处理数据的输入输出，好比：把不一样数据源(RDD，json，Parquet等)获取到数据，并将查询结果输出到DataFrame。
  • catalyst模块：处理SQL语句的整个过程，包括解析，绑定，优化，物理计划等查询优化。
  • hive模块：对hive数据进行处理。
  • hive-ThriftServer：提供CLI以及JDBC和ODBC接口。

3 Spark SQL catalyst模块设计思路

（详细请参看个人SparkSQL源码解析内容）

catalyst主要组件有数据库

  • sqlParse => sql语句的语法解析
  • Analyzer => 将不一样来源的Unresolved Logical Plan和元数据（如hive metastore、Schema catalog）进行绑定，生成resolved Logical Plan
  • optimizer => 根据OptimizationRules，对resolvedLogicalPlan进行合并、列裁剪、过滤器下推等优化做业而转换成optimized Logical Plan
  • Planner => LogicalPlan转换成PhysicalPlan
  • CostModel => 根据过去的性能统计数据，选择最佳的物理执行计划

4 Hash Join的衍生(剑走偏锋)

4.1 Hash join 设计思路剖析(总领全局)

  • 第一步：通常状况下，streamIter为大表，buildIter为小表，不用关心哪一个表为streamIter，哪一个表为buildIter，这个spark会根据join语句自动帮咱们完成。
  • 第二步：根据buildIter Table的join key构建Hash Table，把每一行记录都存进HashTable，位于内存中。
  • 第三步：扫描streamIter Table 每一行数据，使用相同的hash函数匹配 Hash Table中的记录，匹配成功以后再检查join key 是否相等，最后join在一块儿
  • 总结 : hash join 只扫描两表一次，能够认为运算复杂度为o(a+b)，效率很是高。笛卡尔集运算复杂度为a*b。另外，构建的Hash Table最好能所有加载在内存，效率最高，这就决定了hash join算法只适合至少一个小表的join场景，对于两个大表的join场景并不适用。

4.2 broadcast Hash join 设计思路剖析（大表join极小表）

  • 第一步：通常状况下，streamIter为大表，buildIter为小表，不用关心哪一个表为streamIter，哪一个表为buildIter，这个spark会根据join语句自动帮咱们完成。json

  • 第二步： 先把小表广播到全部大表分区所在节点，而后根据buildIter Table的join key构建Hash Table，把每一行记录都存进HashTable微信

  • 第三步：扫描streamIter Table 每一行数据，使用相同的hash函数匹配 Hash Table中的记录，匹配成功以后再检查join key 是否相等，最后join在一块儿分布式

  • 总结 : hash join 只扫描两表一次，能够认为运算复杂度为o(a+b)。函数

  • 调优性能

    1 buildIter整体估计大小超过spark.sql.autoBroadcastJoinThreshold设定的值，
  即不知足broadcast join条件
  
 2 开启尝试使用hash join的开关，spark.sql.join.preferSortMergeJoin=false
  
 3 每一个分区的平均大小不超过spark.sql.autoBroadcastJoinThreshold设定的值，即shuffle read阶段每一个分区来自buildIter的记录要能放到内存中
  
 4 streamIter的大小是buildIter三倍以上
复制代码

4.2 shuffle Hash join 设计思路剖析（大表join小表）

  • 第一步：通常状况下，streamIter为大表，buildIter为小表，不用关心哪一个表为streamIter，哪一个表为buildIter，这个spark会根据join语句自动帮咱们完成。
  • 第二步： 将具备相同性质的（如Hash值相同）join key 进行Shuffle到同一个分区。
  • 第三步：先把小表广播到全部大表分区所在节点，而后根据buildIter Table的join key构建Hash Table，把每一行记录都存进HashTable
  • 第四步：扫描streamIter Table 每一行数据，使用相同的hash函数匹配 Hash Table中的记录，匹配成功以后再检查join key 是否相等，最后join在一块儿

5 Sort Merge join (横行无敌)（大表join大表）

  • 第一步：通常状况下，streamIter为大表，buildIter为小表，不用关心哪一个表为streamIter，哪一个表为buildIter，这个spark会根据join语句自动帮咱们完成。
  • 第二步： 将具备相同性质的（如Hash值相同）join key 进行Shuffle到同一个分区。
  • 第三步： 对streamIter 和 buildIter在shuffle read过程当中先排序，join匹配时按顺序查找，匹配结束后没必要重头开始，利用shuffle sort特性，查找性能解决了大表对大表的情形。

6 Spark Join 类型详解

6.0 准备数据集( Justin => 左表有，Rose =>右表有)

学习 Python中单引号，双引号，3个单引号及3个双引号的区别请参考：https://blog.csdn.net/woainishifu/article/details/76105667

from pyspark.sql.types import * 

>>> rdd1 = sc.parallelize([(1,'Alice', 18),(2,'Andy', 19),(3,'Bob', 17),(4,'Justin', 21),(5,'Cindy', 20)]
park.createDataFrame(rdd, schema)
df.show()>>> schema = StructType([ StructField("id", IntegerType(), True), StructField("name", StringType(), True), StructField("age", IntegerType(), True) ]) 
>>> df = spark.createDataFrame(rdd, schema)
>>> df.show()

+---+------+---+
| id|  name|age|
+---+------+---+
|  1| Alice| 18|
|  2|  Andy| 19|
|  3|   Bob| 17|
|  4|Justin| 21|
|  5| Cindy| 20|
+---+------+---+

>>> rdd2 = sc.parallelize([('Alice', 160),('Andy', 159),('Bob', 170),('Cindy', 165),('Rose', 160)]) 
show()>>> schema2 = StructType([ StructField("name", StringType(), True), StructField("height", IntegerType(), True) ]) 
>>> df2 = spark.createDataFrame(rdd2, schema2) 
>>> df2.show()
+-----+------+
| name|height|
+-----+------+
|Alice|   160|
| Andy|   159|
|  Bob|   170|
|Cindy|   165|
| Rose|   160|
+-----+------+
复制代码

6.1 inner join

  • inner join是必定要找到左右表中知足join key 条件的记录，join key都存在的情形。学习

    df.join(df2, "name", "inner").select("id", df.name, "age", "height").orderBy("id").show()
  df.join(df3, ["id", "name"], "inner").select(df.id, df.name,"age", "height").orderBy(df.id).show()
  df.join(df3, ["id", "name"], "inner").select(df.id, df['name'],"age", "height").orderBy(df.id).show()
  
   >>> df.join(df2, "name", "inner").select("id", df.name, "age", "height").orderBy("id").show()
      +---+-----+---+------+
      | id| name|age|height|
      +---+-----+---+------+
      |  1|Alice| 18|   160|
      |  2| Andy| 19|   159|
      |  3|  Bob| 17|   170|
      |  5|Cindy| 20|   165|
      +---+-----+---+------+   
复制代码

6.2 left outer join

  • left outer join是以左表为准，在右表中查找匹配的记录，若是查找失败，左表行Row不变，右表一行Row中全部字段都为null的记录。大数据

  • 要求：左表是streamIter，右表是buildIter

    df.join(df2, "name", "left").select("id", df.name, "age", "height").orderBy("id").show()
      
  >>> df.join(df2, "name", "left").select("id", "name", "age", "height").orderBy("id").show()
      +---+------+---+------+
      | id|  name|age|height|
      +---+------+---+------+
      |  1| Alice| 18|   160|
      |  2|  Andy| 19|   159|
      |  3|   Bob| 17|   170|
      |  4|Justin| 21|  null|
      |  5| Cindy| 20|   165|
      +---+------+---+------+
复制代码

6.3 right outer join

  • right outer join是以右表为准，在左表中查找匹配的记录，若是查找失败，右表行Row不变，左表一行Row中全部字段都为null的记录。

  • 要求：右表是streamIter，左表是buildIter

    df.join(df2, "name", "right").select("id", df2.name, "age", "height").orderBy("id").show()
  >>> df.join(df2, "name", "right").select("id", "name", "age", "height").orderBy("id").show()
  +----+-----+----+------+
  |  id| name| age|height|
  +----+-----+----+------+
  |null| Rose|null|   160|
  |   1|Alice|  18|   160|
  |   2| Andy|  19|   159|
  |   3|  Bob|  17|   170|
  |   5|Cindy|  20|   165|
  +----+-----+----+------+
复制代码

6.4 full outer join

  • full outer join仅采用sort merge join实现，左边和右表既要做为streamIter，又要做为buildIter

  • 左表和右表已经排好序，首先分别顺序取出左表和右表中的一条记录，比较key，若是key相等，则joinrowA和rowB，并将rowA和rowB分别更新到左表和右表的下一条记录。

  • 若是keyA<keyB，说明右表中没有与左表rowA对应的记录，那么joinrowA与nullRow。

  • 将rowA更新到左表的下一条记录；若是keyA>keyB，则说明左表中没有与右表rowB对应的记录，那么joinnullRow与rowB。

  • 将rowB更新到右表的下一条记录。如此循环遍历直到左表和右表的记录所有处理完。 

    >>> df.join(df2, "name", "outer").select("id", "name", "age", "height").orderBy("id").show()
      +----+------+----+------+
      |  id|  name| age|height|
      +----+------+----+------+
      |null|  Rose|null|   160|
      |   1| Alice|  18|   160|
      |   2|  Andy|  19|   159|
      |   3|   Bob|  17|   170|
      |   4|Justin|  21|  null|
      |   5| Cindy|  20|   165|
      +----+------+----+------+
复制代码

6.5 left semi join

left semi join是以左表为准，在右表中查找匹配的记录，若是查找成功，则仅返回左表Row的记录，不然返回null。

6.6 left anti join

left anti join与left semi join相反，是以左表为准，在右表中查找匹配的记录，若是查找成功，则返回null，不然仅返回左边的记录

6.6 row_number().over()

from pyspark.sql.types import *
from pyspark.sql import Window
from pyspark.sql.functions import *
rdd = sc.parallelize([(1,'Alice', 18),(2,'Andy', 19),(3,'Bob', 17),(1,'Justin', 21),(1,'Cindy', 20)])
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])

df = spark.createDataFrame(rdd, schema)
df.withColumn("rn", row_number().over(Window.partitionBy("id").orderBy("age"))).show()
     +---+------+---+---+
    | id|  name|age| rn|
    +---+------+---+---+
    |  1| Alice| 18|  1|
    |  1| Cindy| 20|  2|
    |  1|Justin| 21|  3|
    |  3|   Bob| 17|  1|
    |  2|  Andy| 19|  1|
    +---+------+---+---+

df.withColumn("rn", row_number().over(Window.partitionBy("id").orderBy("age"))).orderBy("age").show()

    +---+------+---+---+
    | id|  name|age| rn|
    +---+------+---+---+
    |  3|   Bob| 17|  1|
    |  1| Alice| 18|  1|
    |  2|  Andy| 19|  1|
    |  1| Cindy| 20|  2|
    |  1|Justin| 21|  3|
    +---+------+---+---+
复制代码

7 结语

一直想深刻挖掘一下SparkSQL内部join原理，终于有时间详细的理一下 Shuffle Join 。做者还准备进一步研究Spark SQL 内核原理，敬请期待个人Spark SQL源码剖析系列。大数据商业实战社区微信公众号即将开启，敬请关注，谢谢！

在这里插入图片描述
秦凯新 于深圳 201811200130
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程