暂停Flink KeyedStream上的处理

Question

我有一个Flink流应用程序，需要能够“暂停”和“取消暂停”对特定键控流的处理。“处理”意味着只是在流上执行一些简单的异常检测。

我们正在考虑的flow是这样工作的：

命令流，可以是ProcessCommand、PauseCommand或ResumeCommand，每个命令都有一个用于按键的id。

处理命令将检查按键在处理前是否暂停，如果没有暂停，则检查缓冲区。

暂停命令（PauseCommands）将暂停按键处理。

恢复命令（ResumeCommands）将取消对键的暂停处理并刷新缓冲区。

这个流程看起来合理吗？如果合理，我是否可以使用类似拆分操作符的方法来实现？

示例流，输入的单个记录时间戳：

<代码>[{命令：进程，id:1}，{命令：暂停，id:1}，{命令：进程，id:1}，{命令：恢复，id:1}，{命令：进程，id:1}]

Flow:
=>
{command: process, id: 1} # Sent downstream for analysis
=> 
{command: pause, id: 1} # Starts the buffer for id 1
=>
{command: process, id: 1} # Buffered into another output stream
=> 
{command: resume, id: 1} # Turns off buffering, flushes [{command: process, id: 1}] downstream
=>
{command: process, id: 1} # Sent downstream for processing as the buffering has been toggled off 

Answer 1

这可以通过使用Flink的窗口操作符来实现。首先，通过应用映射操作创建基于POJO或元组的流。

然后，根据您的需要，您可以在该流上使用keyBy来获取keyedStream。

现在，通过结合使用基于时间的无限窗口、触发器和窗口函数，可以实现命令流的切换行为。

基本上，您可以使用windows作为缓冲区，它在接收到暂停记录后，会保存进程记录，直到接收到恢复记录。您将编写一个自定义触发器，根据您的场景逐出窗口（缓冲区）。

下面是具有重写方法的触发器的自定义实现。

/**
 * We trigger the window processing as per command inside the record. The
 * process records are buffered when a pause record is received and the
 * buffer is evicted once resume record is received. If no pause record is
 * received earlier, then for each process record the buffer is evicted.
 */
@Override
public TriggerResult onElement(Tuple2<Integer, String> element, long timestamp, Window window,
        TriggerContext context) throws Exception {
    if (element.f1.equals("pause")) {
        paused = true;
        return TriggerResult.CONTINUE;
    } else if (element.f1.equals("resume")) {
        paused = false;
        return TriggerResult.FIRE_AND_PURGE;
    } else if (paused) // paused is a ValueState per keyed stream.
        return TriggerResult.CONTINUE;
    return TriggerResult.FIRE_AND_PURGE;
}

查看此github存储库中的完整工作示例