如果您的流是按时间排序的（只需要为每个单独的设备对流进行排序），那么您可以轻松地转换流以使分析更容易。一 RichFlatMapFunction 这样会将开关事件序列转换为状态更改事件序列：

static class DetectChanges extends RichFlatMapFunction<String, String> {
    private transient ValueState<String> previousState;

    @Override
    public void open(Configuration parameters) throws Exception {
        previousState = getRuntimeContext().getState(new ValueStateDescriptor<>("previousState", String.class));
    }

    @Override
    public void flatMap(String onOrOff, Collector<String> out) throws Exception {

        if (previousState.value() != onOrOff) {
            out.collect("CHANGE");
            previousState.update(onOrOff);
        }
    }
}

现在问题已经简化为确定流在一段时间间隔内是否有一些更改事件。这可以很容易地完成滑动窗口，或者你可以使用cep如果你喜欢。
你也可以完全用cep来做。从概念上讲，您可以按以下方式处理此问题：
定义一个与on+off匹配的单独模式+
然后定义一个模式组，每当在某个时间间隔内出现n次时，该模式组就与该开/关模式匹配