仔细回忆了下最初为啥要改造实现组合 eventTimeTrigger 和 continuousProcessTimeTrigger。还是因为我的watermark是latestTs-10小时，我是考虑如果出现了数据延迟10小时左右的情况，那么我恢复的时候，数据补充进来，如果使用 continuousEventTimeTrigger，10个小时的数据快速涌入，会快速导致很多此event time trigger。相当于单key单window，很快触发 10小时/10s =3600次，全key全window肯定会爆炸。 考虑到这种情况下，我期望基于processTime进行continuous trigger，所以做了这种改造。
yidan zhao <[email protected]> 于2022年6月28日周二 12:25写道： > > 刚刚贴完代码，就分析除问题来了，如下。 > 我看了下，变化主要是 ContinuousProcessingTimeTrigger 中的注册 trigger 时的时间逻辑，加了个 > long nextFireTimestamp = Math.min(time + interval, > window.maxTimestamp()); 取min的逻辑。 > 我这个任务的watermark是latestTs-10小时，因为这个任务特别，压力不大，但对数据完整性要求高，综合考虑这么做的。 > 同时，为了避免10小时后才输出结果，而且需求上，窗口闭合前就需要输出结果，每10s一次，因此采用 continuousProcessTrigger > 这种方式。 > > 我分析，对于同一个key下的窗口1，窗口闭合时trigger触发，然后触发下一次，然后会在同一个time继续registerTime，然后立即触发，死循环，这个过程对于每个key的每个窗口都需要持续10小时，因为窗口需要10小时后才会闭合。 > > > > 不清楚加这么个逻辑的目的是什么呢？ 对于ContinuousProcessingTimeTrigger来说，是基于pt进行触发的，而 > window.maxTimetamp() 在我的场景下是 et。 > 找了下https://issues.apache.org/jira/browse/FLINK-20443对到这个jira，没看懂最终讨论了个啥结论，这貌似也不像是啥bug，为啥需要这么改呢。 > > yidan zhao <[email protected]> 于2022年6月28日周二 11:48写道： > > > > 对比实验了下，就是自定义的 trigger > > 问题，不清楚从1.13.2到1.13.6有啥不同。我的自定义trigger如下，内部就是组装了EventTimeTrigger和ContinuousProcessingTimeTrigger： > > > > public class ContinuousProcessTimeTriggerForEventTimeWindow > > extends Trigger<Object, TimeWindow> { > > > > private final EventTimeTrigger eventTimeTrigger; > > > > private final ContinuousProcessingTimeTrigger<TimeWindow> > > continuousProcessTimeTrigger; > > > > public static ContinuousProcessTimeTriggerForEventTimeWindow > > of(long windowUpdateTimeInSeconds) { > > return new > > ContinuousProcessTimeTriggerForEventTimeWindow(windowUpdateTimeInSeconds); > > } > > > > private ContinuousProcessTimeTriggerForEventTimeWindow(long > > windowUpdateTimeInSeconds) { > > eventTimeTrigger = EventTimeTrigger.create(); > > continuousProcessTimeTrigger = ContinuousProcessingTimeTrigger.of( > > Time.seconds(windowUpdateTimeInSeconds) > > ); > > } > > > > @Override > > public TriggerResult onElement( > > Object element, long timestamp, TimeWindow window, > > TriggerContext ctx > > ) throws Exception { > > continuousProcessTimeTrigger.onElement(element, timestamp, window, > > ctx); > > return eventTimeTrigger.onElement(element, timestamp, window, ctx); > > } > > > > @Override > > public TriggerResult onEventTime( > > long time, TimeWindow window, TriggerContext ctx > > ) throws Exception { > > return eventTimeTrigger.onEventTime(time, window, ctx); > > } > > > > @Override > > public TriggerResult onProcessingTime( > > long time, TimeWindow window, TriggerContext ctx > > ) throws Exception { > > return continuousProcessTimeTrigger.onProcessingTime(time, window, > > ctx); > > } > > > > @Override > > public void clear( > > TimeWindow window, TriggerContext ctx > > ) throws Exception { > > eventTimeTrigger.clear(window, ctx); > > continuousProcessTimeTrigger.clear(window, ctx); > > } > > > > @Override > > public boolean canMerge() { > > return true; > > } > > > > @Override > > public void onMerge( > > TimeWindow window, OnMergeContext ctx > > ) throws Exception { > > eventTimeTrigger.onMerge(window, ctx); > > continuousProcessTimeTrigger.onMerge(window, ctx); > > } > > } > > > > Shengkai Fang <[email protected]> 于2022年6月28日周二 10:51写道： > > > > > > Hi. > > > > > > 这种情况下可以用 jprofile 看看到底 cpu 花在哪里。你可以使用火焰图或者 jstack 看看具体的栈和使用。 > > > > > > Best, > > > Shengkai > > > > > > yidan zhao <[email protected]> 于2022年6月28日周二 10:44写道： > > > > > > > 目前现象如题。任务就是kafkaSource读取数据，简单过滤，然后window，然后输出到mysql。 > > > > > > > > 目前来看运行后1-2min后cpu开始异常，不是马上异常。 异常时候window算子busy为100%。 > > > > window是event time window，配合自定义的 > > > > continuousProcessTriggerForEventTimeWindow（基于pt进行continuous > > > > trigger，但是统计窗口是et window）。 > > > > > > > > 请问这种怎么排查呢？目前来看应该是卡在某个地方了，cancel任务后，直接等到tm失败。window算子百分百不会cancel成功。 > > > >
