ToB企服应用市场:ToB评测及商务社交产业平台

标题: Spark Bloom Filter Join [打印本页]

---

作者: 张春    时间: 2024-9-13 05:20
标题: Spark Bloom Filter Join
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企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4