public class MyProcessFunction extends KeyedProcessFunction<String, Tuple3<String, Long, Integer>, String> {

    private static final int THRESHOLD = 100;
    // 窗口开始（当天 0 点）之后 28 小时，即次日凌晨 4 点清理
    private static final long CLEAR_INTERVAL = 28 * 60 * 60 * 1000L;

    // 窗口聚合状态：<窗口开始时间戳, 聚合值>
    private transient MapState<Long, Integer> aggState;

    @Override
    public void open(Configuration config) throws Exception {
        MapStateDescriptor<Long, Integer> descriptor =
                new MapStateDescriptor<>("Aggregation", Long.class, Integer.class);
        aggState = getRuntimeContext().getMapState(descriptor);
    }

    @Override
    public void processElement(Tuple3<String, Long, Integer> value, Context ctx, Collector<String> out)
            throws Exception {
        // 状态聚合：按事件所在的那一天累加
        long windowStart = getWindowStartWithOffset(value.f1);
        Integer agg = value.f2;
        if (aggState.contains(windowStart)) {
            agg += aggState.get(windowStart);
        }
        aggState.put(windowStart, agg);

        // 阈值判断
        if (agg > THRESHOLD) {
            out.collect("Alert Happens!");
        }

        // 状态清理：为这一天注册一个清理定时器（同一时间点重复注册只会触发一次）
        ctx.timerService().registerProcessingTimeTimer(windowStart + CLEAR_INTERVAL);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        aggState.remove(timestamp - CLEAR_INTERVAL);
    }

    /** 东八区的当天 0 点。 */
    private long getWindowStartWithOffset(long timestamp) {
        long offset = 8 * 60 * 60 * 1000L;
        long windowSize = 24 * 60 * 60 * 1000L;
        long remainder = (timestamp + offset) % windowSize;
        return remainder < 0 ? timestamp - (remainder + windowSize) : timestamp - remainder;
    }
}

dataStream.keyBy(value -> value.f0)
    // 按天滚动，偏移 -8 小时对齐东八区的 0 点
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
    .trigger(new Trigger<Tuple3<String, Long, Integer>, TimeWindow>() {
        private static final int THRESHOLD = 100;
        private final ValueStateDescriptor<Integer> stateDesc =
                new ValueStateDescriptor<>("agg", Integer.class);

        @Override
        public TriggerResult onElement(Tuple3<String, Long, Integer> element, long timestamp,
                TimeWindow window, TriggerContext ctx) throws Exception {
            // 更新累计值
            ValueState<Integer> aggState = ctx.getPartitionedState(stateDesc);
            int agg = element.f2 + (aggState.value() == null ? 0 : aggState.value());
            aggState.update(agg);
            // 超过阈值立即触发，而不是等到窗口结束
            return agg >= THRESHOLD ? TriggerResult.FIRE : TriggerResult.CONTINUE;
        }

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

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

        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            ctx.getPartitionedState(stateDesc).clear();
        }
    })
    // ReduceFunction 只保留一个元素，而不是窗口里的全部元素
    .reduce((value1, value2) -> value2,
        new ProcessWindowFunction<Tuple3<String, Long, Integer>, String, String, TimeWindow>() {
            @Override
            public void process(String key, Context context,
                    Iterable<Tuple3<String, Long, Integer>> elements, Collector<String> out) {
                out.collect("Alert Happens");
            }
        });

-- 状态保留 2 天
SET 'table.exec.state.ttl' = '2 d';

SELECT
    DATE_FORMAT(`time`, 'yyyyMMdd') AS `window_start`,
    `key_word` AS `key`,
    MAX(`time`) AS `time`,
    SUM(`key1`) AS `metric`
FROM `dwd_log`
GROUP BY DATE_FORMAT(`time`, 'yyyyMMdd'), `key_word`
HAVING SUM(`key1`) > 100;

-- 自动把 COUNT DISTINCT 拆成两阶段聚合，缓解倾斜
SET 'table.optimizer.distinct-agg.split.enabled' = 'true';

SELECT
    DATE_FORMAT(`window_start`, 'yyyy-MM-dd HH:mm:ss') AS `window_start`,
    DATE_FORMAT(`window_end`, 'yyyy-MM-dd HH:mm:ss') AS `window_end`,
    `key_word` AS `dim`,
    COUNT(DISTINCT `key2`) AS `metric`
FROM TABLE(
    -- 每分钟输出一次「当天 0 点到现在」的累计值
    CUMULATE(TABLE `dwd_log`, DESCRIPTOR(`rowtime`), INTERVAL '1' MINUTES, INTERVAL '1' DAYS))
GROUP BY `key_word`, `window_start`, `window_end`;

class DemoUDF(ScalarFunction):

    def open(self, function_context: FunctionContext):
        model_path = "your model path"
        script_path = "your python script path"

        with open(script_path, "r") as f:
            code = f.read()

        # 让用户脚本里定义的函数进入全局命名空间
        exec(code, globals())

        self.model = user_open(model_path)

    def eval(self, value):
        return user_eval(self.model, value)