静态网站模板古典,域名访问网站 过程,网店代运营一年的费用是多少,鹤山区网站建设本套技术专栏是作者#xff08;秦凯新#xff09;平时工作的总结和升华#xff0c;通过从真实商业环境抽取案例进行总结和分享#xff0c;并给出商业应用的调优建议和集群环境容量规划等内容#xff0c;请持续关注本套博客。版权声明#xff1a;禁止转载#xff0c;欢迎…本套技术专栏是作者秦凯新平时工作的总结和升华通过从真实商业环境抽取案例进行总结和分享并给出商业应用的调优建议和集群环境容量规划等内容请持续关注本套博客。版权声明禁止转载欢迎学习。QQ邮箱地址1120746959qq.com如有任何商业交流可随时联系。 1 Spark SQL 坚实后盾DataFrame DataFrame是一个分布式数据容器更像传统数据库的二维表格除了数据以外还掌握数据的结构信息即schema。同时与Hive类似DataFrame也支持嵌套数据类型struct、array和map。JSON schema自动推导Hive风格分区表自动识别充分利用RCFile、ORC、Parquet等列式存储格式的优势仅扫描查询真正涉及的列忽略其余列的数据。聚合统计函数支持2 Spark SQL 源码包结构(溯本逐源) 主要分为4类 core模块处理数据的输入输出比如把不同数据源(RDDjsonParquet等)获取到数据并将查询结果输出到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 Planoptimizer 根据OptimizationRules对resolvedLogicalPlan进行合并、列裁剪、过滤器下推等优化作业而转换成optimized Logical PlanPlanner LogicalPlan转换成PhysicalPlanCostModel 根据过去的性能统计数据选择最佳的物理执行计划 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(ab)效率非常高。笛卡尔集运算复杂度为a*b。另外构建的Hash Table最好能全部加载在内存效率最高这就决定了hash join算法只适合至少一个小表的join场景对于两个大表的join场景并不适用。4.2 broadcast Hash join 设计思路剖析大表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(ab)。 调优 1 buildIter总体估计大小超过spark.sql.autoBroadcastJoinThreshold设定的值即不满足broadcast join条件2 开启尝试使用hash join的开关spark.sql.join.preferSortMergeJoinfalse3 每个分区的平均大小不超过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/76105667from 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分别更新到左表和右表的下一条记录。 如果keyAkeyB说明右表中没有与左表rowA对应的记录那么joinrowA与nullRow。 将rowA更新到左表的下一条记录如果keyAkeyB则说明左表中没有与右表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
