火花流口水-性能

Question

我在Scala/Spark中有一个批处理作业，它根据一些输入动态创建Drools规则，然后评估规则。我还有一个与要插入到规则引擎的事实相对应的输入。

到目前为止，我正在一个接一个地插入事实，然后触发关于这个事实的所有规则。我正在使用rdd.aggregate执行此操作。

seqOp运算符的定义如下：

scala prettyprint-override">/**
 * @param broadcastRules the broadcasted KieBase object containing all rules
 * @param aggregator used to accumulate values when rule matches
 * @param item the fact to run Drools with
 * @tparam T the type of the given item
 * @return the updated aggregator
 */
def seqOp[T: ClassTag](broadcastRules: Broadcast[KieBase])(
  aggregator: MyAggregator,
  item: T) : MyAggregator = {
  val session = broadcastRules.value.newStatelessKieSession
  session.setGlobal("aggregator", aggregator)
  session.execute(CommandFactory.newInsert(item))
  aggregator
}

以下是生成的规则的示例：

dialect "mvel"
global batch.model.MyAggregator aggregator
rule "1"
 when condition
 then do something on the aggregator
end 

对于同一RDD，该批次花了20分钟来评估3K规则，但花了10小时来评估10K规则！

我想知道根据事实插入事实是否是最好的方法。一次插入RDD的所有项目比触发所有规则更好吗？对我来说，这似乎不是最佳选择，因为所有事实都将同时存在于工作记忆中。

您认为上面的代码有任何问题吗？

Answer 1

最后我发现了这个问题，它更多地与规则匹配时在聚合器上执行的操作有关，而不是与规则的评估有关。