您好,登录后才能下订单哦!
本篇内容主要讲解“spark的动态分区裁剪怎么实现”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“spark的动态分区裁剪怎么实现”吧!
本文基于delta 0.7.0 spark 3.0.1 spark 3.x引入了动态分区裁剪
直接定位到PartitionPruning.applyPartitionPruning是逻辑计划的规则
override def apply(plan: LogicalPlan): LogicalPlan = plan match { // Do not rewrite subqueries. case s: Subquery if s.correlated => plan case _ if !SQLConf.get.dynamicPartitionPruningEnabled => plan case _ => prune(plan) }
当是该逻辑计划是子查询且该子查询是相关的,则直接跳过,因为相关的子查询将会被重写到join条件中
如果没有开启动态分区,则直接跳过
其他条件则会跳到下一步 下一步的条件,则是会判断是否是包含join操作,如果是join操作才会进行后续的操作:
private def prune(plan: LogicalPlan): LogicalPlan = { plan transformUp { // skip this rule if there's already a DPP subquery on the LHS of a join case j @ Join(Filter(_: DynamicPruningSubquery, _), _, _, _, _) => j case j @ Join(_, Filter(_: DynamicPruningSubquery, _), _, _, _) => j case j @ Join(left, right, joinType, Some(condition), hint) =>
具体分析一下每一步: 1.
var newLeft = left var newRight = right // extract the left and right keys of the join condition val (leftKeys, rightKeys) = j match { case ExtractEquiJoinKeys(_, lkeys, rkeys, _, _, _, _) => (lkeys, rkeys) case _ => (Nil, Nil) } //ExtractEquiJoinKeys的unapply方法 def unapply(join: Join): Option[ReturnType] = join match { case Join(left, right, joinType, condition, hint) => logDebug(s"Considering join on: $condition") // Find equi-join predicates that can be evaluated before the join, and thus can be used // as join keys. val predicates = condition.map(splitConjunctivePredicates).getOrElse(Nil) val joinKeys = predicates.flatMap { case EqualTo(l, r) if l.references.isEmpty || r.references.isEmpty => None case EqualTo(l, r) if canEvaluate(l, left) && canEvaluate(r, right) => Some((l, r)) case EqualTo(l, r) if canEvaluate(l, right) && canEvaluate(r, left) => Some((r, l)) // Replace null with default value for joining key, then those rows with null in it could // be joined together case EqualNullSafe(l, r) if canEvaluate(l, left) && canEvaluate(r, right) => Seq((Coalesce(Seq(l, Literal.default(l.dataType))), Coalesce(Seq(r, Literal.default(r.dataType)))), (IsNull(l), IsNull(r)) ) case EqualNullSafe(l, r) if canEvaluate(l, right) && canEvaluate(r, left) => Seq((Coalesce(Seq(r, Literal.default(r.dataType))), Coalesce(Seq(l, Literal.default(l.dataType)))), (IsNull(r), IsNull(l)) ) case other => None }
ExtractEquiJoinKeys用来提取and条件分隔的多个条件,之后只有条件满足相等的才能进行下一步处理:
如果相等但是左边或者右边的表达式的为空,则不匹配
如果相等而且有对应的逻辑计划能够产生对应的属性值,则匹配
如果是EqualNullsafe,且有相应的逻辑能够产生相应的属性值,则会转换为Coalesce和isnull的判断
之后转化为leftKeys和rightKeys表达式 如join的条件是:tableA.a1 = tableB.b2 AND tableA.a2=tableB.b2
则经过该过程得到的结果为leftKey为:Seq(tableA.a1,tableA.a2) rightKeys为:Seq(tableB.b1,tableB.b2)
splitConjunctivePredicates(condition).foreach { case EqualTo(a: Expression, b: Expression) if fromDifferentSides(a, b) => val (l, r) = if (a.references.subsetOf(left.outputSet) && b.references.subsetOf(right.outputSet)) { a -> b } else { b -> a } // there should be a partitioned table and a filter on the dimension table, // otherwise the pruning will not trigger var partScan = getPartitionTableScan(l, left) if (partScan.isDefined && canPruneLeft(joinType) && hasPartitionPruningFilter(right)) { val hasBenefit = pruningHasBenefit(l, partScan.get, r, right) newLeft = insertPredicate(l, newLeft, r, right, rightKeys, hasBenefit) } else { partScan = getPartitionTableScan(r, right) if (partScan.isDefined && canPruneRight(joinType) && hasPartitionPruningFilter(left) ) { val hasBenefit = pruningHasBenefit(r, partScan.get, l, left) newRight = insertPredicate(r, newRight, l, left, leftKeys, hasBenefit) } } case _ => }
对每一个Equals对,先对左边表达式进行getPartitionTableScan 操作,该方法的作用是:
找到该表达式的最终逻辑计划,并且返回
只有该逻辑计划是HadoopFsRelation类型且存在partition列的时候
,才返回该逻辑计划
如果join左边逻辑计划满足getPartitionTableScan,且join的类型是innerjoin/leftSemi/RightOuter,且该join右边逻辑计划不是一个流且存在比如> <这种的filter, 才会在左边逻辑计划插入一个DynamicPruningSubquery的父节点,但是插入该节点还有两个条件是pruningHasBenefit或者SQLConf.get.exchangeReuseEnabled 满足,默认SQLConf.get.exchangeReuseEnabled是ture
对于右边的逻辑计划也是类似的处理方式。只不过join的类型要求为inner/LeftOuter
pruningHasBenefit方法的计算逻辑为: 如果filterRatio*getPartitionTableScan.stats.sizeInByte>该逻辑计划涉及的所有叶子节点.stats.sizeInByte 则可以添加DynamicPruningSubquery
返回整个新的join操作
Join(newLeft, newRight, joinType, Some(condition), hint
到此,相信大家对“spark的动态分区裁剪怎么实现”有了更深的了解,不妨来实际操作一番吧!这里是亿速云网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。