滑动时间窗口的Flink性能问题

Question

我正在尝试flink的一些网络监控工作。我的目标是计算每个src\u ip的不同dst\u ip。

我下面的代码工作，但性能真的很糟糕。似乎每个滑动窗口都重新计算所有事件，但这不应该是必要的。

例如，我们有时间秒1-600的事件。Flink可以得到每秒的累加器，所以我们每秒有600个累加器。当第一个滑动窗口过期时，flink只合并1-300的累加器，并销毁第二个1的累加器。此窗口还可以在最后一秒前预合并1-299。当第二个滑动窗口过期时，flink只合并2-301的累加器，并销毁第二个2的累加器。等等

这种方法比将事件分配给多个窗口和计算每个窗口的聚合要有效得多。

Flink支持这一点吗？我能自己用flink得到类似的函数吗？

谢谢！

public static class AverageAccumulator2 {
    String key;
    Set<String> target;
    AverageAccumulator2() {
        target = new HashSet<>();
    }
}

public static class Average2 implements AggregateFunction<ObjectNode, AverageAccumulator2, Tuple3<String, Long, Set<String>>> {
    @Override
    public AverageAccumulator2 createAccumulator() {
        return new AverageAccumulator2();
    }

    @Override
    public AverageAccumulator2 add(ObjectNode value, AverageAccumulator2 accumulator) {
        accumulator.key = value.get("value").get("src_ip").asText();
        accumulator.target.add(value.get("value").get("dst_ip").asText());
        return accumulator;
    }
    @Override
    public Tuple3<String, Long, Set<String>> getResult(AverageAccumulator2 accumulator) {
        return new Tuple3<>(accumulator.key, (long) accumulator.target.size(), accumulator.target);
    }

    @Override
    public AverageAccumulator2 merge(AverageAccumulator2 a, AverageAccumulator2 b) {
        a.target.addAll(b.target);
        return a;
    }
}

final SingleOutputStreamOperator<Tuple3<String, Long, Set<String>> > process2 =
stream.keyBy(value -> value.get("value").get("sip").asText())
                    .timeWindow(Time.seconds(300),Time.seconds(1))
                    .aggregate(new Average2());

Answer 1

正如您所观察到的，Flink不会尝试优化滑动窗口。对于细粒度滑动，这确实会变得非常昂贵。

您可以做的是使用ProcessFunction实现您自己的逻辑来处理状态和定时器——您可以像您概述的那样实现它。您将拥有对于每个传入记录更新用于累积结果的数据结构的过程元素方法，以及每秒触发一次、将部分结果合并在一起并向下游发送结果的onTimer方法。