Java Flink 창 트리거 사용 방법

정의

Trigger는 창(창 할당자에 의해 형성됨)이 창 기능에 의해 처리될 준비가 되는 시기를 결정합니다. 각 WindowAssigner에는 기본값인 Trigger가 함께 제공됩니다. 기본 트리거가 요구 사항을 충족하지 않는 경우 Trigger(…)를 사용할 수 있습니다.

트리거 소스 코드

public abstract class Trigger<T, W extends Window> implements Serializable {
	/**
	 只要有元素落⼊到当前窗⼝， 就会调⽤该⽅法
	 * @param element 收到的元素
	 * @param timestamp 元素抵达时间.
	 * @param window 元素所属的window窗口.
	 * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调.
	 */
    public abstract TriggerResult onElement(T var1, long var2, W var4, Trigger.TriggerContext var5) throws Exception;
	
	 /**
	 * processing-time 定时器回调函数
	 *
	 * @param time 定时器触发的时间.
	 * @param window 定时器触发的窗口对象.
	 * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调.
	 */
    public abstract TriggerResult onProcessingTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception;

	/**
	 * event-time 定时器回调函数
	 *
	 * @param time 定时器触发的时间.
	 * @param window 定时器触发的窗口对象.
	 * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调.
	 */
    public abstract TriggerResult onEventTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception;

	 /**
	 * 当 多个窗口合并到⼀个窗⼝的时候，调用该方法法，例如系统SessionWindow
	 *
	 * @param window 合并后的新窗口对象
	 * @param ctx ⼀个上下⽂对象，通常用该对象注册 timer（ProcessingTime/EventTime）回调以及访问状态
	 */
    public void onMerge(W window, Trigger.OnMergeContext ctx) throws Exception {
        throw new UnsupportedOperationException("This trigger does not support merging.");
    }
	
	/**
	 * 当窗口被删除后执⾏所需的任何操作。例如：可以清除定时器或者删除状态数据
	 */
    public abstract void clear(W var1, Trigger.TriggerContext var2) throws Exception;
    }

TriggerResult 소스 코드

public enum TriggerResult {
	// 表示对窗口不执行任何操作。即不触发窗口计算，也不删除元素。
    CONTINUE(false, false),
    // 触发窗口计算，输出结果，然后将窗口中的数据和窗口进行清除。
    FIRE_AND_PURGE(true, true),
    // 触发窗口计算，但是保留窗口元素
    FIRE(true, false),
    // 不触发窗口计算，丢弃窗口，并且删除窗口的元素。
    PURGE(false, true);

    private final boolean fire;
    private final boolean purge;

    private TriggerResult(boolean fire, boolean purge) {
        this.purge = purge;
        this.fire = fire;
    }

    public boolean isFire() {
        return this.fire;
    }

    public boolean isPurge() {
        return this.purge;
    }
}

트리거가 창이 처리할 준비가 되었다고 판단하면 실행되며 반환 상태는 FIRE 또는 FIRE_AND_PURGE일 수 있습니다. 그중 FIRE는 창 계산을 트리거하고 창 내용을 유지하는 반면, FIRE_AND_PURGE는 창 계산을 트리거하고 창 내용을 삭제합니다. 기본적으로 사전 구현된 트리거는 창 상태를 지우지 않고 단순히 FIRE됩니다.

Flink의 사전 설정 Trigger

• EventTimeTrigger: EventTime과 Window Endtime을 비교하여 창 계산을 트리거할지 여부를 결정합니다. 그렇지 않으면 트리거되지 않습니다. 창은 계속 기다릴 것입니다.

• ProcessTimeTrigger: ProcessTime과 창 EndTme를 비교하여 창을 트리거할지 여부를 결정합니다. ProcessTime이 EndTime보다 크면 계산이 트리거되고, 그렇지 않으면 창이 계속 대기합니다.

• ContinuousEventTimeTrigger: 간격 주기적 트리거 창을 기준으로 계산되거나 창의 종료 시간이 현재 EndTime 트리거 창보다 작습니다.

• ContinuousProcessingTimeTrigger: 간격 주기적 트리거 창을 기준으로 계산되거나 창의 종료 시간이 현재 ProcessTime 트리거 창보다 작습니다.

• CountTrigger: 액세스된 데이터의 양이 설정된 임계값을 초과하는지 여부에 따라 창 계산을 트리거할지 여부를 결정합니다.

• DeltaTrigger: 액세스 데이터에서 계산된 델타 표시기가 지정된 임계값을 초과하는지 여부에 따라 창 계산을 트리거할지 여부를 결정합니다.

• PurgingTrigger: 모든 트리거를 매개변수로 Purge 유형 트리거로 변환할 수 있으며, 계산이 완료된 후 데이터가 정리됩니다.

• NeverTrigger: 언제든지 창 계산을 트리거하지 않습니다.

주로 EventTimeTrigger 및 ProcessTimeTrigger의 소스 코드를 살펴보세요.

EventTimeTrigger 소스 코드

public class EventTimeTrigger extends Trigger<Object, TimeWindow> {
    private static final long serialVersionUID = 1L;

    private EventTimeTrigger() {
    }

    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            return TriggerResult.FIRE;
        } else {
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }
    }

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

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

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

    public boolean canMerge() {
        return true;
    }

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

    }

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

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

ProcessingTimeTrigger 소스 코드

public class ProcessingTimeTrigger extends Trigger<Object, TimeWindow> {
    private static final long serialVersionUID = 1L;

    private ProcessingTimeTrigger() {
    }

    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
        ctx.registerProcessingTimeTimer(window.maxTimestamp());
        return TriggerResult.CONTINUE;
    }

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

    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
        return TriggerResult.FIRE;
    }

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

    public boolean canMerge() {
        return true;
    }

    public void onMerge(TimeWindow window, OnMergeContext ctx) {
        long windowMaxTimestamp = window.maxTimestamp();
        if (windowMaxTimestamp > ctx.getCurrentProcessingTime()) {
            ctx.registerProcessingTimeTimer(windowMaxTimestamp);
        }

    }

    public String toString() {
        return "ProcessingTimeTrigger()";
    }

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

onElement() 메소드에서 ctx.registerProcessingTimer(window.maxTimestamp())는 ProcessTime 타이머를 등록하고 시간 매개변수는 window.maxTimestamp()입니다. 창의 마지막 시간, 시간이 이 창의 마지막 시간에 도달하면 타이머가 onProcessingTime() 메서드를 트리거하고 호출합니다. onProcessingTime() 메서드에서 TriggerResult.FIRE를 반환하면 창의 데이터 계산이 트리거됩니다. , 그러나 창 요소는 유지합니다.

ProcessingTimeTrigger 클래스는 창의 마지막 시간이 도달할 때만 창 함수 계산을 트리거합니다. 계산이 완료된 후에는 창의 데이터가 메모리에 저장되지 않습니다. PURGE 또는 FIRE_AND_PURGE가 호출되지 않는 한 데이터는 항상 메모리에 있습니다. 실제로 PurgingTrigger 클래스를 제외하고 Flink에서 제공하는 Trigger 클래스 중 어느 것도 창의 데이터를 지우지 않습니다.

공용 창에 대한 트리거

스크롤 창

TumblingEventTimeWindows ：EventTimeTrigger
public class TumblingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
            return EventTimeTrigger.create();
        }
}

TumblingProcessingTimeWindows: ProcessingTimeTrigger

public class TumblingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return ProcessingTimeTrigger.create();
    }
}

슬라이딩 창

SlidingEventTimeWindows：EventTimeTrigger
public class SlidingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return EventTimeTrigger.create();
    }
}

SlidingProcessingTimeWindows: ProcessingTimeTrigger

public class SlidingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
            return ProcessingTimeTrigger.create();
        }
}

세션 창

EventTimeSessionWindows：EventTimeTrigger
public class EventTimeSessionWindows extends MergingWindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return EventTimeTrigger.create();
    }
}

ProcessingTimeSessionWindows : ProcessTimeTrigger

public class ProcessingTimeSessionWindows extends MergingWindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return ProcessingTimeTrigger.create();
    }
}

전역 창

GlobalWindows ：NeverTrigger
public class GlobalWindows extends WindowAssigner<Object, GlobalWindow> {
     public Trigger<Object, GlobalWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
            return new GlobalWindows.NeverTrigger();
        }
}

위 내용은 Java Flink 창 트리거 사용 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!