Flink内存不足

我有一些相当简单的流代码，它们通过时间窗口聚合数据。窗口偏大（1小时，限制为2小时），流中的值是来自数百个服务器的度量。我的内存不足，因此添加了RocksDBStateBackend。这导致JVM出现段错误。接下来，我尝试了FsStateBackend。这两个后端都从未将任何数据写入磁盘，而只是使用JobID创建了一个目录。我在独立模式下（未部署）运行此代码。是否对为什么State Backends不写数据以及为什么即使有8GB的堆却耗尽了内存有什么想法？

    final SingleOutputStreamOperator<Metric> metricStream =
            objectStream.map(node -> new Metric(node.get("_ts").asLong(), node.get("_value").asDouble(), node.get("tags"))).name("metric stream");

    final WindowedStream<Metric, String, TimeWindow> hourlyMetricStream = metricStream
            .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Metric>(Time.hours(2)) { // set how long metrics can come late
                @Override
                public long extractTimestamp(final Metric metric) {
                    return metric.get_ts() * 1000; // needs to be in ms since Java epoch
                }
            })
            .keyBy(metric -> metric.getMetricName()) // key the stream so we can run the windowing in parallel
            .timeWindow(Time.hours(1)); // setup the time window for the bucket

    // create a stream for each type of aggregation
    hourlyMetricStream.sum("_value") // we want to sum by the _value
    .addSink(new MetricStoreSinkFunction(parameters, "sum"))
    .name("hourly sum stream")
    .setParallelism(6);

    hourlyMetricStream.aggregate(new MeanAggregator())
    .addSink(new MetricStoreSinkFunction(parameters, "mean"))
    .name("hourly mean stream")
    .setParallelism(6);

    hourlyMetricStream.aggregate(new ReMedianAggregator())
    .addSink(new MetricStoreSinkFunction(parameters, "remedian"))
    .name("hourly remedian stream")
    .setParallelism(6);

    env.execute("flink test");