0

我想创建一个Trigger第一次在 20 秒内触发，之后每五秒触发一次。我用过GlobalWindows和一个习惯Trigger

val windowedStream = valueStream
                          .keyBy(0)
                          .window(GlobalWindows.create())
                          .trigger(TradeTrigger.of())

这是中的代码TradeTrigger：

@PublicEvolving
public class TradeTrigger<W extends Window> extends Trigger<Object, W> {

    private static final long serialVersionUID = 1L;

    static boolean flag=false;
    static long ctime = System.currentTimeMillis();

    private TradeTrigger() {
    }

    @Override
    public TriggerResult onElement(
            Object arg0,
            long arg1,
            W arg2,
            org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext arg3)
            throws Exception {
        // TODO Auto-generated method stub

        if(flag == false){
            if((System.currentTimeMillis()-ctime) >= 20000){
               flag = true;
               ctime = System.currentTimeMillis();
               return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        } else {
            if((System.currentTimeMillis()-ctime) >= 5000){
                ctime = System.currentTimeMillis();
                return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        }

    }

    @Override
    public TriggerResult onEventTime(
            long arg0,
            W arg1,
            org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext arg2)
            throws Exception {
        // TODO Auto-generated method stub
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(
            long arg0,
            W arg1,
            org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext arg2)
            throws Exception {
        // TODO Auto-generated method stub
        return TriggerResult.CONTINUE;
    }


    public static <W extends Window> TradeTrigger<W> of() {
        return new TradeTrigger<>();
    }

}

所以基本上，when flagis false，即第一次，Trigger应该在 20 秒内触发并将 设置flag为true。从下一次开始，它应该每 5 秒触发一次。

我面临的问题是，每次Trigger触发时我只会在输出中收到一条消息。也就是说，我在 20 秒后收到一条消息，每五秒收到一条消息。我预计每次触发时输出中有 20 条消息。

如果我使用.timeWindow(Time.seconds(5))并创建一个 5 秒的时间窗口，则每 5 秒输出 20 条消息。请帮助我正确获取此代码。有什么我想念的吗？

4

2 回答 2

5

您的 Trigger 实现存在一些问题：

  1. 您永远不应该将函数的状态存储在静态变量中。Flink 不隔离 JVM 中的用户进程。相反，它为每个 TaskManager 使用一个 JVM 并启动多个线程。因此，您的静态布尔标志在多个触发器实例之间共享。您应该存储标志 Flink 的ValueState界面，该界面可从TriggerContext. Flink 将注意检查您的状态并在发生故障时恢复它。

  2. Trigger.onEvent()仅在新事件到达时调用。因此它不能用于在特定时间触发 Window 计算。相反，您应该注册一个事件时间计时器或处理时间计时器（再次通过TriggerContext）。计时器将分别调用Trigger.onEventTime()或Trigger.onProcessingTime()。是使用事件还是处理时间取决于您的用例。

于 2016-04-22T11:54:53.077 回答
1

在 Fabian 和 Flink 邮件列表的答案的帮助下得到了它。通过 .将状态存储在ValueState变量中TriggerContext。检查方法中的变量，onEvent()如果是第一次，则注册 aprocessingTimeTimer比当前时间多 20 秒并更新状态。在该onProcessingTime方法中，注册另一个ProcessingTimeTimer比当前时间多 5 秒，更新状态并触发Window.

于 2016-04-25T12:46:15.090 回答