Flink：在数据流和"set of rules"之间实现"join"

Question

以下用例的最佳实践建议是什么？我们需要将流与一组"规则"进行匹配，这些规则本质上是一个 Flink 数据集概念。可以更新此"规则集"，但并不频繁。每个流事件必须根据">规则集"中的所有记录进行检查，并且每个匹配项都会在接收器数据流中生成一个或多个事件。规则集中的记录数在 6 位范围内。

目前，我们只是将规则加载到本地规则列表中，并在传入的数据流上使用flatMap。在flatMap中，我们只是迭代一个列表，将每个事件与每个规则进行比较。

为了加快迭代速度，我们还可以将列表分成几批，本质上是创建一个列表列表，并创建一个单独的线程来迭代每个子列表(使用 Java 或 Scala 中的 Futures(。

问题：

1. 有没有更好的方法来执行这种连接？
2. 如果不是，那么在每个 flatMap 操作中创建新线程来添加额外的并行性是否安全，在 Flink 已经在做的事情之上？

编辑： 以下是请求的示例代码：

package wikiedits
import org.apache.flink.streaming.connectors.wikiedits.{WikipediaEditEvent, WikipediaEditsSource}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.extensions._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
object WikipediaEditEventProcessor {
def main(args: Array[String])= {
val see = StreamExecutionEnvironment.getExecutionEnvironment
val edits = see.addSource(new WikipediaEditsSource())
val ruleSets = Map[Int, List[String]](
(1, List("a", "b", "c", "d", "e", "f", "g", "h", "i", "j")),
(2, List("k", "l", "m", "n", "o", "p", "q", "r", "s", "t")),
(3, List("u", "v", "w", "x", "y", "z", "0", "1", "2", "3"))
)
val result = edits.flatMap { edit =>
ruleSets.map { ruleSet =>
applyRuleSet(edit, ruleSet._2, ruleSet._1)
}
}
see.execute
}
def applyRuleSet(event: WikipediaEditEvent, ruleSet: List[String], ruleSetId: Int): Future[List[String]] = {
val title = event.getTitle
Future(
ruleSet.map {
case rule if title.contains(rule) =>
val result = s"Ruleset $ruleSetId: $rule -> exists in: $title"
println(result) // this would be creating an output event instead
result
case rule =>
val result = s"Ruleset $ruleSetId: $rule -> NO MATCH in: $title"
println(result)
result
}
)
}
}

Answer 1

每个流事件必须根据"规则集"中的所有记录进行检查， 每个匹配项都会在接收器数据流中生成一个或多个事件。 规则集中的记录数在 6 位范围内

假设你有 K 条规则。如果输入速率快于处理单个事件的 K 个规则所花费的时间，则您的方法很好。 否则，您需要一些可以并行处理这些 K 规则的方法。

将它们视为K收费站。将它们一个接一个地放置，而不是将它们放在一个大房间里。这将简化流引擎的事情。

换句话说，使用简单的 for 循环遍历所有规则，并为每个规则提供一个单独的 flatMap。 因此，它们中的每一个都是相互独立的，因此可以并行处理。 最后，您将有K flatMaps用于执行。引擎将使用最大并行性，无论您提供什么配置用于执行。 此方法将最大可能的并行度限制为 K。但是，这对于大量规则来说已经足够了。

通过在每个平面映射内创建新线程来增加并行性 操作

完全不推荐。将并行性留给闪烁。您可以在flatMap中定义要执行的工作单元。