flink聚合状态巨大，如何解决

Question

我尝试用不同的窗口大小计算流中的数据（窗口大小在steam数据中），所以我使用自定义的WindowAssigner和AggregateFunction，但状态很大（窗口范围从一小时到30天）

在我看来，聚合状态只是存储中间结果

有什么问题吗？

public class ElementProcessingTime extends WindowAssigner<Element, TimeWindow> {
    @Override public Collection<TimeWindow> assignWindows(Element element, long timestamp, WindowAssignerContext context) {
        long slide = Time.seconds(10).toMilliseconds();
        long size = element.getTime() * 60 * 1000;
        timestamp = context.getCurrentProcessingTime();

        List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
        long lastStart = TimeWindow.getWindowStartWithOffset(timestamp, 0, slide);
        for (long start = lastStart; start > timestamp - size; start -= slide) {
            windows.add(new TimeWindow(start, start + size));
        }
        return windows;
    }

    @Override public Trigger<FactorCalDetail, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return ElementTimeTrigger.create();
    }

    @Override public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
        return new TimeWindow.Serializer();
    }

    @Override public boolean isEventTime() {
        return false;
    }
}

public class CountAggregate implements AggregateFunction<FactorCalDetail, AggregateResult, AggregateResult> {

    @Override public AggregateResult createAccumulator() {
        AggregateResult result = new AggregateResult();
        result.setResult(0.0);
        return result;
    }

    @Override public AggregateResult add(FactorCalDetail value, AggregateResult accumulator) {
        accumulator.setKey(value.getGroupKey());
        accumulator.addResult();
        accumulator.setTimeSpan(value.getTimeSpan());
        return accumulator;
    }

    @Override public AggregateResult getResult(AggregateResult accumulator) {
        return accumulator;
    }

    @Override public AggregateResult merge(AggregateResult a, AggregateResult b) {
        if (a.getKey().equals(b.getKey())) {
            a.setResult(a.getResult() + b.getResult());
        }
        return a;
    }
}

env.addSource(source)
    .keyBy(Element::getKey)
    .window(new ElementProcessingTime())
    .aggregate(new CountAggregate())
    .addSink(new RedisCustomizeSink(redisProperties));

Answer 1

当您分配自定义窗口时，状态大小可能会很快失控。这主要是因为每个窗口都需要保存属于它的所有记录，直到窗口被聚合并最终被驱逐。在您的代码中，您似乎也为每条记录创建了大量窗口。

您没有指定您的用例，但我假设您实际上想要计算每个键在 10 毫秒 bin 大小的给定时间点上有多少事件延伸。如果是这样，那么这不是 windows 的直接用例。

你想做的是：

1. 将您的活动拆分为更小的活动。
2. 按 key 和 bin 分组。
3. 清点你的垃圾箱。

代码中的粗略草图：

input.flatMap(element -> {
        ...
        for (long start = lastStart; start > timestamp - size; start -= slide) {
            emit(new KeyTime(key, start));
        }
    })
    .keyBy(keyTime -> keyTime)
    .count()

您可以在keyBy 之后应用窗口来强制某些输出属性，例如等待几分钟然后输出所有内容并忽略迟到的事件。

注意：KeyTime 是一个简单的 POJO，包含密钥和 bin 时间。

编辑：在您发表评论后，解决方案实际上要简单得多。

env.addSource(source)
    .keyBy(element -> new Tuple2<>(element.getKey(), element.getTime()))
    .count()
    .addSink(new RedisCustomizeSink(redisProperties));

Answer 2

你没有说源是什么，它会有自己的状态持续存在。您也没有说有多少个唯一键。随着唯一键数量的增加，即使每个键的少量状态也会变得巨大。如果问题最终出现在聚合器状态的增长中，您可以尝试将窗口逻辑拆分为一系列两个窗口，一个用于每小时聚合，另一个用于将每小时汇总聚合到您想要的时间范围。