Re: Flink Session Window在窗口创建的时候输出一条数据

Mon, 11 May 2020 20:19:01 -0700 

我想通过trigger的方式来实现这个需求，想法是重写EventTimeTrigger  
,在首条记录进入系统时，触发一下purge，通过ValueStatus 
记录状态。但是现在遇到的问题是，会重复触发多少。而且窗口关闭时的触发，WindowFunction Function收到的记录也不对。请问是我使用方式不对吗？

自定义的trigger：
public class SessionComputeTrigger extends Trigger<SessionConvertBean, 
TimeWindow> {
    private static final long serialVersionUID = 1L;
    static Logger logger = LoggerFactory.getLogger(SessionComputeTrigger.class);
    private final ReducingStateDescriptor<Long> countDesc = new 
ReducingStateDescriptor<>("count", new SessionComputeTrigger.Sum(), 
LongSerializer.INSTANCE);

    private final ValueStateDescriptor<Boolean> existDesc = new 
ValueStateDescriptor<>("exist", BooleanSerializer.INSTANCE);

    private SessionComputeTrigger() {
    }

    @Override
    public TriggerResult onElement(SessionConvertBean element, long timestamp, 
TimeWindow window, TriggerContext ctx) throws Exception {
        ValueState<Boolean> exist = ctx.getPartitionedState(existDesc);
        ReducingState<Long> count = ctx.getPartitionedState(countDesc);

        if (exist == null || exist.value() == null || !exist.value()) {
            count.add(1L);
            logger.info("mau fire =====>exits:{} 
====>count:{}====>date_time:{}", exist.value(), count.get(), element.dateTime);
            exist.update(true);
            return TriggerResult.FIRE;
        }

        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            // if the watermark is already past the window fire immediately
            return TriggerResult.FIRE;
        } else {
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, 
TriggerContext ctx) {
        return time == window.maxTimestamp() ? TriggerResult.FIRE : 
TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, 
TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.deleteEventTimeTimer(window.maxTimestamp());
    }

    @Override
    public boolean canMerge() {
        return true;
    }

    @Override
    public void onMerge(TimeWindow window,
                        OnMergeContext ctx) {
        long windowMaxTimestamp = window.maxTimestamp();
        if (windowMaxTimestamp > ctx.getCurrentWatermark()) {
            ctx.registerEventTimeTimer(windowMaxTimestamp);
        }
    }

    @Override
    public String toString() {
        return "EventTimeTrigger()";
    }

    public static SessionComputeTrigger create() {
        return new SessionComputeTrigger();
    }

    private static class Sum implements ReduceFunction<Long> {
        private static final long serialVersionUID = 1L;

        @Override
        public Long reduce(Long value1, Long value2) throws Exception {
            logger.info("value1={},value2={}", value1, value2);
            return value1 + value2;
        }

    }
}

输出值：

mau fire =====>exits:null 
====>count:1====>date_time:2020-05-10T23:22:40.298+0800  
（date_time是当前记录的eventtime）
  
result=>用户：48b132cb-f40b-4cf5-bf7f-b6411525b1982,开始时间：2020-05-10T23:22:40.298+0800,结束时间：2020-05-10T23:23:10.298+0800,间隔：30000.0,总数：1
mau fire =====>exits:null 
====>count:1====>date_time:2020-05-10T23:22:43.298+0800
 
result=>用户：48b132cb-f40b-4cf5-bf7f-b6411525b1982,开始时间：2020-05-10T23:22:40.298+0800,结束时间：2020-05-10T23:23:13.298+0800,间隔：33000.0,总数：2
mau fire =====>exits:null 
====>count:1====>date_time:2020-05-10T23:22:46.298+0800
 
result=>用户：48b132cb-f40b-4cf5-bf7f-b6411525b1982,开始时间：2020-05-10T23:22:40.298+0800,结束时间：2020-05-10T23:23:16.298+0800,间隔：36000.0,总数：3
mau fire =====>exits:null 
====>count:1====>date_time:2020-05-10T23:22:49.298+0800 



在 2020/5/11 下午4:12，“Benchao 
Li”<user-zh-return-3480-gougang_1991=163....@flink.apache.org 代表 
libenc...@gmail.com> 写入:

    这个暂时应该是没有办法做到这一点。或者你可以用两个query来实现这个？
    比如一个query是统计first_value；第二个是真正的session window。

    gang.gou <gougang_1...@163.com> 于2020年5月11日周一 下午3:07写道：

    > Hi，
    >
    >     我想使用Flink的Session Window
    > 
去统计一个用户的在线访问时长，和统计当前活跃的用户数量，目前只能在window结束的时候才输出一条包含窗口开始和结束时间的记录，怎么在窗口创建的时候就先输出一条记录，结束的时候再去更新这条记录呢，谢谢！
    >
    >

    -- 

    Benchao Li
    School of Electronics Engineering and Computer Science, Peking University
    Tel:+86-15650713730
    Email: libenc...@gmail.com; libenc...@pku.edu.cn

回复