flink IngestionTime介绍

flink的窗口时间属性TimeCharacteristic分为三种：ProcessingTime，IngestionTime，EventTime。

1. ProcessingTime是处理时间，所有基于时间的操作(如时间窗口)将使用运行各自操作符的机器的系统时间，优点是简单，缺点是依赖每个节点的系统时间，如果数据流速不一样比如出现反压等会导致窗口数据不确定；
2. EventTime是事件时间，是从数据里面提取出来的时间，EventTime依赖于数据，不依赖于系统时间，优点是能严格按照数据时间的发生顺序进行窗口统计，缺点是如果数据出现断流了，会导致watermark无法提高，从而无法导致窗口的触发
3. IngestionTime是摄入时间，是数据从数据源取出的时候附带上系统时间作为watermark，作为ProcessingTime和EventTime的折中方案，会定时的往下游发送watermark，这个watermark是系统时间，不会因为数据断流导致watermark无法提高。适用于对数据延迟不大，对数据窗口统计要求不是很严格的场景。
   下面接下来从源码分析IngestionTime，代码在StreamSourceContexts.java中：

switch (timeCharacteristic) {
            case EventTime:
                ctx = new ManualWatermarkContext<>(
                    output,
                    processingTimeService,
                    checkpointLock,
                    streamStatusMaintainer,
                    idleTimeout);

                break;
            case IngestionTime:
                ctx = new AutomaticWatermarkContext<>(
                    output,
                    watermarkInterval,
                    processingTimeService,
                    checkpointLock,
                    streamStatusMaintainer,
                    idleTimeout);

                break;
            case ProcessingTime:
                ctx = new NonTimestampContext<>(checkpointLock, output);
                break;
            default:
                throw new IllegalArgumentException(String.valueOf(timeCharacteristic));
        }

IngestionTime是通过AutomaticWatermarkContext类来实现逻辑的，继续看AutomaticWatermarkContext：

private AutomaticWatermarkContext(
                final Output<StreamRecord<T>> output,
                final long watermarkInterval,
                final ProcessingTimeService timeService,
                final Object checkpointLock,
                final StreamStatusMaintainer streamStatusMaintainer,
                final long idleTimeout) {

            super(timeService, checkpointLock, streamStatusMaintainer, idleTimeout);

            this.output = Preconditions.checkNotNull(output, "The output cannot be null.");

            Preconditions.checkArgument(watermarkInterval >= 1L, "The watermark interval cannot be smaller than 1 ms.");
            this.watermarkInterval = watermarkInterval;

            this.reuse = new StreamRecord<>(null);

            this.lastRecordTime = Long.MIN_VALUE;

            long now = this.timeService.getCurrentProcessingTime();
            //注册定时器，等到下一个watermarkInterval的时候触发
            this.nextWatermarkTimer = this.timeService.registerTimer(now + watermarkInterval,
                new WatermarkEmittingTask(this.timeService, checkpointLock, output));
}

可以看到，最后一行代码那里，注册了一个定时器，在下一个watermarkInterval时触发执行，再看触发的WatermarkEmittingTask里面的逻辑：

            @Override
            public void onProcessingTime(long timestamp) {
                final long currentTime = timeService.getCurrentProcessingTime();

                synchronized (lock) {
                    // we should continue to automatically emit watermarks if we are active
                    if (streamStatusMaintainer.getStreamStatus().isActive()) {
                        if (idleTimeout != -1 && currentTime - lastRecordTime > idleTimeout) {
                            // if we are configured to detect idleness, piggy-back the idle detection check on the
                            // watermark interval, so that we may possibly discover idle sources faster before waiting
                            // for the next idle check to fire
                            markAsTemporarilyIdle();

                            // no need to finish the next check, as we are now idle.
                            cancelNextIdleDetectionTask();
                        } else if (currentTime > nextWatermarkTime) {
                            // align the watermarks across all machines. this will ensure that we
                            // don't have watermarks that creep along at different intervals because
                            // the machine clocks are out of sync
                            // 这里是发送的watermark的值，取整处理
                            final long watermarkTime = currentTime - (currentTime % watermarkInterval);

                            output.emitWatermark(new Watermark(watermarkTime));
                            nextWatermarkTime = watermarkTime + watermarkInterval;
                        }
                    }
                }
                // 注册下一次定时器，下一次的执行时间又是间隔watermarkInterval
                long nextWatermark = currentTime + watermarkInterval;
                nextWatermarkTimer = this.timeService.registerTimer(
                        nextWatermark, new WatermarkEmittingTask(this.timeService, lock, output));
            }

至此，逻辑已经很清楚了，IngestionTime是每经过watermarkInterval间隔发送一次watermark，watermark的值就是当前系统时间取整：currentTime - (currentTime % watermarkInterval)。IngestionTime并不会因为数据断流导致watermark无法提升，如果对数据延迟不大，对数据窗口统计要求不是很严格的场景，同时可能出现数据断流的情况下，IngestionTime比较适用。