是否可以在需要时调用Flink Map（在输入流上未激活）

Question

我有一个闪烁的地图，一旦数据通过流就会被激活。

即使没有数据通过，我也要调用该地图。

我将map移动到一个函数中（无限函数调用），但是flink作业永远不会运行。如果我将其添加到map中，它只会在数据通过时被激活。

想法是，在一个infinte循环中有一个map，检查一些共享变量和另一个闪烁流监控kafka队列，如果数据进入它的过程，它会更改一个共享变量，以某种方式影响无限循环并继续。

如何调用无限循环映射并一起运行flink映射？

我尝试用随机数据创建CollectionMap来激活流和map以调用无限循环，但即使map中存在一个while（true）条件，也几乎立即退出

在IDE中，当我将其推到Flink时，它会工作。本地it几乎立即退出，不停留在循环中

流1

    val data_stream = env.addSource(myConsumer)
      .map(x => {process(x)})

流2

    val elements = List[String]("Start")
    var read = env.fromElements(elements).map(x => ProcessData.infinteLoop())

如何调用无限循环映射并一起运行flink映射？

Answer 1

您可以创建一个窗口和一个触发器，并每隔x秒调用一次映射。

您可以在以下文档中找到：<代码>https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/windows.html

示例：

import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows
import org.apache.flink.streaming.api.windowing.triggers.{CountTrigger, PurgingTrigger}
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow

 val data_stream = env.addSource(myConsumer)
  .map(x => {process(x)})

val window: DataStream[String] = data_stream
  .windowAll(GlobalWindows.create())
  .trigger(PurgingTrigger.of(CountTrigger.of[GlobalWindow](5)))
  .apply((w: GlobalWindow, x: Iterable[(Integer, String)], y: Collector[String]) => {})