Spark Bloom Filter Join

张春  金牌会员 | 2024-9-13 05:20:41 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 916|帖子 916|积分 2748

1 综述

1.1 目的

  Bloom Filter Join，或者说Row-level Runtime Filtering（还额外有一条Semi-Join分支），是Spark 3.3对运行时过滤的一个最新补充
  之前运行时过滤主要有两个：动态分区裁剪DPP（开源实现）、动态文件裁剪DFP（Databricks实现），两者都能有效淘汰数据源层面的Scan IO
  Bloom Filter Join的主要优化点是在shuffle层，通过在join shuffle前对表进行过滤从而提高运行效率
1.2 场景



  • 普通的shuffle join

  • Broadcast join并且子布局中存在shuffle

1.3 底子过程

  将存在过滤条件的小表端称为Filter Creation Side，另一层称为Filter Application Side
  对于如下的SQL：SELECT * FROM R JOIN S ON R.r_sk = S.s_sk where S.x = 5
  起首Creation端进行bloomFilter创建，简朴来说就是对小表创建一个bloomFilter的过滤数据集合
  1. SELECT BloomFilterAggregate(XxHash64(S.s_sk), n_items, n_bits)
  2. FROM S where S.x = 5
复制代码
  之后Application端进行重写（实际是整个查询重写），就是把小表的bloomFilter数据集合拿来对大表的数据进行过滤
  根据上面的场景图看，其实小表Creation端在整个SQL树上并没有变革，只改变了大表端的树布局
  1. SELECT *
  2. FROM R JOIN S ON R.r_sk = S.s_sk
  3. WHERE S.x=5 AND BloomFilterMightContain(
  4. (
  5.   SELECT BloomFilterAggregate(XxHash64(S.s_sk), n_items, n_bits) bloom_filter
  6.                     FROM S where S.x = 5 ),     -- Bloom filter creation
  7.                   XxHash64(R.r_sk))       -- Bloom filter application
复制代码
1.4 触发条件

  设计文档中写的触发条件

  • 小表在broadcast join当中（存疑）
  • 小表有过滤器
  • 小表是Scan (-> Project) -> Filter的建档形式，否则依赖流增加可能延长查询时间
  • 小表是确定性的
  • 大表端有shuffle，小表可以通过shuffl传送bloomFilter结果
  • join的列上没有应用DPP
2 InjectRuntimeFilter

  InjectRuntimeFilter是Spark源码中对应的优化器类，只执行一次（FixedPoint(1)和Once的差异是Once逼迫幂等）
  1. Batch("InjectRuntimeFilter", FixedPoint(1),
  2.   InjectRuntimeFilter) :+
复制代码
  apply中定义了规则的团体流程，前面是两个条件判定
  1. //  相关子查询不支持，相关子查询的子查询结果依赖于主查询，不能应用
  2. case s: Subquery if s.correlated => plan
  3. //  相关的配置开关是否开启
  4. case _ if !conf.runtimeFilterSemiJoinReductionEnabled &&
  5.   !conf.runtimeFilterBloomFilterEnabled => plan
  6. case _ =>
  7.   //  应用优化规则，尝试注入运行时过滤器
  8.   val newPlan = tryInjectRuntimeFilter(plan)
  9.   //  semi join配置未开或者规则应用后无变化，不处理
  10.   if (conf.runtimeFilterSemiJoinReductionEnabled && !plan.fastEquals(newPlan)) {
  11.   //  子查询重写成semi/anti join
  12.     RewritePredicateSubquery(newPlan)
  13.   } else {
  14.     newPlan
  15.   }
复制代码
  相关的设置为，默认bloomFilter开启了，Semi join关闭的
  1. val RUNTIME_FILTER_SEMI_JOIN_REDUCTION_ENABLED =
  2.   buildConf("spark.sql.optimizer.runtimeFilter.semiJoinReduction.enabled")
  3.     .doc("When true and if one side of a shuffle join has a selective predicate, we attempt " +
  4.       "to insert a semi join in the other side to reduce the amount of shuffle data.")
  5.     .version("3.3.0")
  6.     .booleanConf
  7.     .createWithDefault(false)
  8.    
  9. val RUNTIME_BLOOM_FILTER_ENABLED =
  10.   buildConf("spark.sql.optimizer.runtime.bloomFilter.enabled")
  11.     .doc("When true and if one side of a shuffle join has a selective predicate, we attempt " +
  12.       "to insert a bloom filter in the other side to reduce the amount of shuffle data.")
  13.     .version("3.3.0")
  14.     .booleanConf
  15.     .createWithDefault(true)
复制代码
2.1 tryInjectRuntimeFilter

  tryInjectRuntimeFilter使用焦点的处置惩罚流程，尝试应用Runtime Filter，团体代码如下
  1. private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
  2.   var filterCounter = 0
  3.   val numFilterThreshold = conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
  4.   plan transformUp {
  5.     case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _, left, right, hint) =>
  6.       var newLeft = left
  7.       var newRight = right
  8.       (leftKeys, rightKeys).zipped.foreach((l, r) => {
  9.         // Check if:
  10.         // 1. There is already a DPP filter on the key
  11.         // 2. There is already a runtime filter (Bloom filter or IN subquery) on the key
  12.         // 3. The keys are simple cheap expressions
  13.         if (filterCounter < numFilterThreshold &&
  14.           !hasDynamicPruningSubquery(left, right, l, r) &&
  15.           !hasRuntimeFilter(newLeft, newRight, l, r) &&
  16.           isSimpleExpression(l) && isSimpleExpression(r)) {
  17.           val oldLeft = newLeft
  18.           val oldRight = newRight
  19.           if (canPruneLeft(joinType) && filteringHasBenefit(left, right, l, hint)) {
  20.             newLeft = injectFilter(l, newLeft, r, right)
  21.           }
  22.           // Did we actually inject on the left? If not, try on the right
  23.           if (newLeft.fastEquals(oldLeft) && canPruneRight(joinType) &&
  24.             filteringHasBenefit(right, left, r, hint)) {
  25.             newRight = injectFilter(r, newRight, l, left)
  26.           }
  27.           if (!newLeft.fastEquals(oldLeft) || !newRight.fastEquals(oldRight)) {
  28.             filterCounter = filterCounter + 1
  29.           }
  30.         }
  31.       })
  32.       join.withNewChildren(Seq(newLeft, newRight))
  33.   }
  34. }
复制代码
  过程中有很多的条件判定，应用Runtime Filter的基本条件：

  • 插入的Runtime Filter没凌驾阈值（默认10）
  • 等值条件的Key上不能有DPP、Runtime Filter
  • 等值条件的Key是一个简朴表达式（即没有套上UDF等）
  之后根据条件，选择将Runtime Filter应用到左子树照旧右子树，条件为

  • Join范例支持下推（好比RightOuter只能用于左子树）
  • Application端支持通过joins、aggregates、windows下推过滤条件
  • Creation端有过滤条件
  • 当前join是shuffle join或者是一个子布局中包罗shuffle的broadcast join
  • Application端的扫描数据大于阈值（默认10G）
  提到的两个阈值的设置项
  1. val RUNTIME_FILTER_NUMBER_THRESHOLD =
  2.   buildConf("spark.sql.optimizer.runtimeFilter.number.threshold")
  3.     .doc("The total number of injected runtime filters (non-DPP) for a single " +
  4.       "query. This is to prevent driver OOMs with too many Bloom filters.")
  5.     .version("3.3.0")
  6.     .intConf
  7.     .checkValue(threshold => threshold >= 0, "The threshold should be >= 0")
  8.     .createWithDefault(10)
  10. val RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD =
  11.   buildConf("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold")
  12.     .doc("Byte size threshold of the Bloom filter application side plan's aggregated scan " +
  13.       "size. Aggregated scan byte size of the Bloom filter application side needs to be over " +
  14.       "this value to inject a bloom filter.")
  15.     .version("3.3.0")
  16.     .bytesConf(ByteUnit.BYTE)
  17.     .createWithDefaultString("10GB")
复制代码
2.2 injectFilter

  injectFilter是焦点进行Runtime Filter规则应用的地方，在此处，bloomFilter和Semi Join是互斥的，只能有一个执行
  1. if (conf.runtimeFilterBloomFilterEnabled) {
  2.   injectBloomFilter(
  3.     filterApplicationSideExp,
  4.     filterApplicationSidePlan,
  5.     filterCreationSideExp,
  6.     filterCreationSidePlan
  7.   )
  8. } else {
  9.   injectInSubqueryFilter(
  10.     filterApplicationSideExp,
  11.     filterApplicationSidePlan,
  12.     filterCreationSideExp,
  13.     filterCreationSidePlan
  14.   )
复制代码
2.3 injectBloomFilter

2.3.1 执行条件

  起首进行一个判定，在Creation端的数据不能大于阈值（Creation端数据量大会导致bloomFilter的误判率高，最终过滤效果差）
  1. // Skip if the filter creation side is too big
  2. if (filterCreationSidePlan.stats.sizeInBytes > conf.runtimeFilterCreationSideThreshold) {
  3.   return filterApplicationSidePlan
  4. }
复制代码
  阈值设置默认10M
  1. val RUNTIME_BLOOM_FILTER_CREATION_SIDE_THRESHOLD =
  2.   buildConf("spark.sql.optimizer.runtime.bloomFilter.creationSideThreshold")
  3.     .doc("Size threshold of the bloom filter creation side plan. Estimated size needs to be " +
  4.       "under this value to try to inject bloom filter.")
  5.     .version("3.3.0")
  6.     .bytesConf(ByteUnit.BYTE)
  7.     .createWithDefaultString("10MB")
复制代码
  Creation端的数据是一个预估数据，是LogicalPlan中的属性LogicalPlanStats获取的，分是否开启CBO，详细获取方式待研究
  1. def stats: Statistics = statsCache.getOrElse {
  2.   if (conf.cboEnabled) {
  3.     statsCache = Option(BasicStatsPlanVisitor.visit(self))
  4.   } else {
  5.     statsCache = Option(SizeInBytesOnlyStatsPlanVisitor.visit(self))
  6.   }
  7.   statsCache.get
  8. }
复制代码
2.3.2 创建Creation端的聚合

  就是创建一个bloomFilter的聚合函数BloomFilterAggregate，是AggregateFunction的子类，属于Expression。根据统计信息中是否存在行数，会传入不同的参数
  1. val rowCount = filterCreationSidePlan.stats.rowCount
  2. val bloomFilterAgg =
  3.   if (rowCount.isDefined && rowCount.get.longValue > 0L) {
  4.     new BloomFilterAggregate(new XxHash64(Seq(filterCreationSideExp)), rowCount.get.longValue)
  5.   } else {
  6.     new BloomFilterAggregate(new XxHash64(Seq(filterCreationSideExp)))
  7.   }
复制代码
2.3.3 创建Application端的过滤条件

  根据1.3中的形貌，此处就是把上节中Creation端创建的bloomFilter过滤条件构建成Application端的条件
  Alias就是一个别名的效果；ColumnPruning就是进行列裁剪，后续不必要的列不读取；ConstantFolding就是进行常量折叠；ScalarSubquery是标量子查询，标量子查询的查询结果是一行一列的值（单一值）
  BloomFilterMightContain就是一个内部标量函数，查抄数据是否由bloomFilter包罗，继承自Predicate，返回boolean值
  1. val alias = Alias(bloomFilterAgg.toAggregateExpression(), "bloomFilter")()
  2. val aggregate =
  3.   ConstantFolding(ColumnPruning(Aggregate(Nil, Seq(alias), filterCreationSidePlan)))
  4. val bloomFilterSubquery = ScalarSubquery(aggregate, Nil)
  5. val filter = BloomFilterMightContain(bloomFilterSubquery,
  6.   new XxHash64(Seq(filterApplicationSideExp)))
复制代码
  最终结果是在原Application端的计划树上加一个filter，如下就是最终的返回结果
  1. Filter(filter, filterApplicationSidePlan)
复制代码
2.4 injectInSubqueryFilter

  injectInSubqueryFilter团体流程与injectBloomFilter差不多，差异应该是在Application端天生的过滤条件变成in
  1. val actualFilterKeyExpr = mayWrapWithHash(filterCreationSideExp)
  2. val alias = Alias(actualFilterKeyExpr, actualFilterKeyExpr.toString)()
  3. val aggregate =
  4.   ColumnPruning(Aggregate(Seq(filterCreationSideExp), Seq(alias), filterCreationSidePlan))
  5. if (!canBroadcastBySize(aggregate, conf)) {
  6.   // Skip the InSubquery filter if the size of `aggregate` is beyond broadcast join threshold,
  7.   // i.e., the semi-join will be a shuffled join, which is not worthwhile.
  8.   return filterApplicationSidePlan
  9. }
  10. val filter = InSubquery(Seq(mayWrapWithHash(filterApplicationSideExp)),
  11.   ListQuery(aggregate, childOutputs = aggregate.output))
  12. Filter(filter, filterApplicationSidePlan)
复制代码
  这里有一个小优化就是mayWrapWithHash，当数据范例的大小凌驾int时，就是把数据转为hash
  1. // Wraps `expr` with a hash function if its byte size is larger than an integer.
  2. private def mayWrapWithHash(expr: Expression): Expression = {
  3.   if (expr.dataType.defaultSize > IntegerType.defaultSize) {
  4.     new Murmur3Hash(Seq(expr))
  5.   } else {
  6.     expr
  7.   }
  8. }
复制代码
3 BloomFilterAggregate

  类有三个焦点参数：

  • child：子表达式，就是InjectRuntimeFilter里传的XxHash64，目前看起来数据先经过XxHash64处置惩罚成long再放入BloomFilter
  • estimatedNumItemsExpression：估计的数据量，假如InjectRuntimeFilter没拿到统计信息，就用设置的默认值
  • numBitsExpression：要使用的bit数
  1. case class BloomFilterAggregate(
  2.     child: Expression,
  3.     estimatedNumItemsExpression: Expression,
  4.     numBitsExpression: Expression,
复制代码
  estimatedNumItemsExpression和numBitsExpression对应的设置如下
  1. val RUNTIME_BLOOM_FILTER_EXPECTED_NUM_ITEMS =
  2.   buildConf("spark.sql.optimizer.runtime.bloomFilter.expectedNumItems")
  3.     .doc("The default number of expected items for the runtime bloomfilter")
  4.     .version("3.3.0")
  5.     .longConf
  6.     .createWithDefault(1000000L)
  7.    
  8. val RUNTIME_BLOOM_FILTER_NUM_BITS =
  9.   buildConf("spark.sql.optimizer.runtime.bloomFilter.numBits")
  10.     .doc("The default number of bits to use for the runtime bloom filter")
  11.     .version("3.3.0")
  12.     .longConf
  13.     .createWithDefault(8388608L)
复制代码
  BloomFilter用的是Spark本身实现的一个类BloomFilterImpl，BloomFilterAggregate的createAggregationBuffer接口中创建
  1. override def createAggregationBuffer(): BloomFilter = {
  2.   BloomFilter.create(estimatedNumItems, numBits)
  3. }
复制代码
  参数就是前面的estimatedNumItemsExpression和numBitsExpression，是懒加载的参数（应该在处置惩罚过程会被改变，所以实际跟前面的值之间还加了一层与默认值的比较赋值）
  1. // Mark as lazy so that `estimatedNumItems` is not evaluated during tree transformation.
  2. private lazy val estimatedNumItems: Long =
  3.   Math.min(estimatedNumItemsExpression.eval().asInstanceOf[Number].longValue,
  4.     SQLConf.get.getConf(RUNTIME_BLOOM_FILTER_MAX_NUM_ITEMS))
复制代码
  处置惩罚数据的接口应该是update，把数据用XxHash64处置惩罚后到场BloomFilter
  1. override def update(buffer: BloomFilter, inputRow: InternalRow): BloomFilter = {
  2.   val value = child.eval(inputRow)
  3.   // Ignore null values.
  4.   if (value == null) {
  5.     return buffer
  6.   }
  7.   buffer.putLong(value.asInstanceOf[Long])
  8.   buffer
  9. }
复制代码
  对象BloomFilterAggregate有对应的序列化和反序列化接口
  1. object BloomFilterAggregate {
  2.   final def serialize(obj: BloomFilter): Array[Byte] = {
  3.     // BloomFilterImpl.writeTo() writes 2 integers (version number and num hash functions), hence
  4.     // the +8
  5.     val size = (obj.bitSize() / 8) + 8
  6.     require(size <= Integer.MAX_VALUE, s"actual number of bits is too large $size")
  7.     val out = new ByteArrayOutputStream(size.intValue())
  8.     obj.writeTo(out)
  9.     out.close()
  10.     out.toByteArray
  11.   }
  13.   final def deserialize(bytes: Array[Byte]): BloomFilter = {
  14.     val in = new ByteArrayInputStream(bytes)
  15.     val bloomFilter = BloomFilter.readFrom(in)
  16.     in.close()
  17.     bloomFilter
  18.   }
  19. }
复制代码
4 BloomFilterMightContain

  有两个参数

  • bloomFilterExpression：是上节BloomFilter的二进制数据
  • valueExpression：应该跟上节的child同等，对输入数据做处置惩罚的表达式，XxHash64
  1. case class BloomFilterMightContain(
  2.     bloomFilterExpression: Expression,
  3.     valueExpression: Expression)
复制代码
  bloomFilter通过反序列化获取
  1. // The bloom filter created from `bloomFilterExpression`.
  2. @transient private lazy val bloomFilter = {
  3.   val bytes = bloomFilterExpression.eval().asInstanceOf[Array[Byte]]
  4.   if (bytes == null) null else deserialize(bytes)
  5. }
复制代码
  做数据判定的应该是eval，就是调用的BloomFilter的接口进行判定。eval应该就是Spark中Expression表达式的执行接口
  1. override def eval(input: InternalRow): Any = {
  2.   if (bloomFilter == null) {
  3.     null
  4.   } else {
  5.     val value = valueExpression.eval(input)
  6.     if (value == null) null else bloomFilter.mightContainLong(value.asInstanceOf[Long])
  7.   }
  8. }
复制代码
  也有doGenCode接口用来天生代码
  1. override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
  2.   if (bloomFilter == null) {
  3.     ev.copy(isNull = TrueLiteral, value = JavaCode.defaultLiteral(dataType))
  4.   } else {
  5.     val bf = ctx.addReferenceObj("bloomFilter", bloomFilter, classOf[BloomFilter].getName)
  6.     val valueEval = valueExpression.genCode(ctx)
  7.     ev.copy(code = code"""
  8.     ${valueEval.code}
  9.     boolean ${ev.isNull} = ${valueEval.isNull};
  10.     ${CodeGenerator.javaType(dataType)} ${ev.value} = ${CodeGenerator.defaultValue(dataType)};
  11.     if (!${ev.isNull}) {
  12.       ${ev.value} = $bf.mightContainLong((Long)${valueEval.value});
  13.     }""")
  14.   }
  15. }
复制代码
5 计划变更

  取Spark单位测试的样例（InjectRuntimeFilterSuite）：select * from bf1 join bf2 on bf1.c1 = bf2.c2 where bf2.a2 = 62


  • 规则前的plan
  1. GlobalLimit 21
  2. +- LocalLimit 21
  3.    +- Project [cast(a1#38430 as string) AS a1#38468, cast(b1#38431 as string) AS b1#38469, cast(c1#38432 as string) AS c1#38470, cast(d1#38433 as string) AS d1#38471, cast(e1#38434 as string) AS e1#38472, cast(f1#38435 as string) AS f1#38473, cast(a2#38436 as string) AS a2#38474, cast(b2#38437 as string) AS b2#38475, cast(c2#38438 as string) AS c2#38476, cast(d2#38439 as string) AS d2#38477, cast(e2#38440 as string) AS e2#38478, cast(f2#38441 as string) AS f2#38479]
  4.       +- Join Inner, (c1#38432 = c2#38438)
  5.          :- Filter isnotnull(c1#38432)
  6.          :  +- Relation spark_catalog.default.bf1[a1#38430,b1#38431,c1#38432,d1#38433,e1#38434,f1#38435] parquet
  7.          +- Filter ((isnotnull(a2#38436) AND (a2#38436 = 62)) AND isnotnull(c2#38438))
  8.             +- Relation spark_catalog.default.bf2[a2#38436,b2#38437,c2#38438,d2#38439,e2#38440,f2#38441] parquet
复制代码


  • 规则后的plan
  1. GlobalLimit 21
  2. +- LocalLimit 21
  3.    +- Project [cast(a1#38430 as string) AS a1#38468, cast(b1#38431 as string) AS b1#38469, cast(c1#38432 as string) AS c1#38470, cast(d1#38433 as string) AS d1#38471, cast(e1#38434 as string) AS e1#38472, cast(f1#38435 as string) AS f1#38473, cast(a2#38436 as string) AS a2#38474, cast(b2#38437 as string) AS b2#38475, cast(c2#38438 as string) AS c2#38476, cast(d2#38439 as string) AS d2#38477, cast(e2#38440 as string) AS e2#38478, cast(f2#38441 as string) AS f2#38479]
  4.       +- Join Inner, (c1#38432 = c2#38438)
  5.          :- Filter might_contain(scalar-subquery#38494 [], xxhash64(c1#38432, 42))
  6.          :  :  +- Aggregate [bloom_filter_agg(xxhash64(c2#38438, 42), 1000000, 8388608, 0, 0) AS bloomFilter#38493]
  7.          :  :     +- Project [c2#38438]
  8.          :  :        +- Filter ((isnotnull(a2#38436) AND (a2#38436 = 62)) AND isnotnull(c2#38438))
  9.          :  :           +- Relation spark_catalog.default.bf2[a2#38436,b2#38437,c2#38438,d2#38439,e2#38440,f2#38441] parquet
  10.          :  +- Filter isnotnull(c1#38432)
  11.          :     +- Relation spark_catalog.default.bf1[a1#38430,b1#38431,c1#38432,d1#38433,e1#38434,f1#38435] parquet
  12.          +- Filter ((isnotnull(a2#38436) AND (a2#38436 = 62)) AND isnotnull(c2#38438))
  13.             +- Relation spark_catalog.default.bf2[a2#38436,b2#38437,c2#38438,d2#38439,e2#38440,f2#38441] parquet
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

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

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

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

本版积分规则

张春

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

楼主热帖

标签云

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