简单流程功能的忙碌时间太长答案

【问题标题】：Busy time is too high for simple process function简单流程功能的忙碌时间太长
【发布时间】：2023-01-13 09:52:12
【问题描述】：

我正在使用 flink v1.13，有 4 个任务管理器（每 16 个 cpu）和 3800 个任务（默认应用程序并行度为 28）

在我的应用程序中，一位操作员的忙碌时间总是很高（大约 %80 - %90）。

如果我重新启动 flink 应用程序，那么繁忙时间会减少，但在运行 5-10 小时后，繁忙时间会再次增加。

在 grafana 中，我可以看到 ProcessStream 的繁忙时间增加了。这是 PromethuesQuery：avg((avg_over_time(flink_taskmanager_job_task_busyTimeMsPerSecond[1m]))) by (task_name)

ProcessStream 任务中没有背压。为了计算 backPressure 时间，我使用：flink_taskmanager_job_task_backPressuredTimeMsPerSecond

但我找不到任何理由。

这是代码：

private void processOne(DataStream<KafkaObject> kafkaLog) {
    kafkaLog
         .filter(new FilterRequest())
         .name(FilterRequest.class.getSimpleName())
         .map(new MapToUserIdAndTimeStampMs())
         .name(MapToUserIdAndTimeStampMs.class.getSimpleName())
         .keyBy(UserObject::getUserId) // returns of type int
         .process(new ProcessStream())
         .name(ProcessStream.class.getSimpleName())
         .addSink(...)
         
        ;
}

// ...
// ...

public class ProcessStream extends KeyedProcessFunction<Integer, UserObject, Output>
{
    private static final long STATE_TIMER = // 5 min in milliseconds;

    private static final int AVERAGE_REQUEST = 74;
    private static final int STANDARD_DEVIATION = 32;
    private static final int MINIMUM_REQUEST = 50;
    private static final int THRESHOLD = 70;


    private transient ValueState<Tuple2<Integer, Integer>> state;

    @Override
    public void open(Configuration parameters) throws Exception
    {
        ValueStateDescriptor<Tuple2<Integer, Integer>> stateDescriptor = new ValueStateDescriptor<Tuple2<Integer, Integer>>(
                ProcessStream.class.getSimpleName(),
                TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {}));

        state = getRuntimeContext().getState(stateDescriptor);
    }

    @Override
    public void processElement(UserObject value, KeyedProcessFunction<Integer, UserObject, Output>.Context ctx, Collector<Output> out) throws Exception
    {
        Tuple2<Integer, Integer> stateValue = state.value();

        if (Objects.isNull(stateValue)) {
            stateValue = Tuple2.of(1, 0);
            ctx.timerService().registerProcessingTimeTimer(value.getTimestampMs() + STATE_TIMER);
        }

        int totalRequest = stateValue.f0;
        int currentScore = stateValue.f1;

        if (totalRequest >= MINIMUM_REQUEST && currentScore >= THRESHOLD)
        {
            out.collect({convert_to_output});
            state.clear();
        }
        else
        {
            stateValue.f0 = totalRequest + 1;
            stateValue.f1 =  calculateNextScore(stateValue.f0);
            state.update(stateValue);
        }
    }

    private int calculateNextScore(int totalRequest)
    {
        return (totalRequest - AVERAGE_REQUEST ) / STANDARD_DEVIATION;
    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Integer, UserObject, Output>.OnTimerContext ctx, Collector<Output> out) throws Exception
    {
        state.clear();
    }
}

【问题讨论】：

您在工作流中使用事件时间还是处理时间？如果是事件时间（基于 UserObject.getTimestampMs()），那么您想使用 . registerEventTimeTimer() 而不是 . registerProcessingTimeTimer() 注册计时器。
@kkrugler，我正在使用处理时间
我看到 CPU 随着时间的推移而增加的一种情况是当状态填满 TM 内存时，并且当您接近满堆时您开始获得大量 GC 活动。

标签： apache-flink flink-streaming

【解决方案1】：

由于您使用的是来自传入记录 (value.getTimestampMs() + STATE_TIMER) 的时间戳值，因此您希望使用事件时间运行，并根据传入记录的时间戳设置水印。否则您不知道计时器何时实际触发，因为记录的时间戳可能与您当前的处理器时间完全不同。

这意味着您还想使用.registerEventTimeTimer()。

如果没有这些更改，您可能会用未清除状态填满 TM 堆，这会导致高 CPU 负载。

【讨论】：