如何在Spark Structured Streaming中计算基于前置行的移动平均值

错误根因 #

Spark Structured Streaming 确实默认禁止在流数据集上使用非时间驱动的窗口函数，核心原因是流数据属于无界数据集，缺少时间维度作为状态过期依据的话，窗口计算对应的状态会无限膨胀，最终触发内存溢出，因此框架层面直接做了限制，你遇到的报错就是这个规则的体现。

可行实现方案 #

你需要的按设备分组、取最近N行计算移动平均的需求，可以通过FlatMapGroupsWithState算子手动维护分组状态实现，这也是Spark官方推荐的流处理自定义状态计算方案，完全支持每条新消息到达时同步输出计算结果。

具体实现步骤 #

1. 定义状态结构
   为每个设备分组维护定长列表，存储最近上报的数值，示例状态类：

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

public class MovingAvgState implements Serializable {
    private List<Double> recentValues = new ArrayList<>();

    public List<Double> getRecentValues() {
        return recentValues;
    }

    public void setRecentValues(List<Double> recentValues) {
        this.recentValues = recentValues;
    }
}

2. 实现状态计算逻辑
   基于你已有的lines数据集继续处理即可：

import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.streaming.GroupState;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.streaming.FlatMapGroupsWithStateFunction;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

// 计算3行移动平均（包含当前行往前数共3条，若需要4条可自行调整长度）
Dataset<Row> movingAvgResult = lines
    // 配置水印，用于自动清理长期无数据的设备状态，避免内存泄漏
    .withWatermark("timestamp", "1 hour")
    // 按设备ID分组
    .groupByKey((MapFunction<Row, String>) row -> row.getAs("item"), Encoders.STRING())
    .flatMapGroupsWithState(
        new FlatMapGroupsWithStateFunction<String, Row, MovingAvgState, Row>() {
            @Override
            public Iterator<Row> call(String deviceId, Iterator<Row> inputs, GroupState<MovingAvgState> state) throws Exception {
                // 初始化状态
                MovingAvgState currentState = state.exists() ? state.get() : new MovingAvgState();
                List<Double> recentValues = currentState.getRecentValues();
                List<Row> outputList = new ArrayList<>();

                while (inputs.hasNext()) {
                    Row currentRow = inputs.next();
                    Double value = currentRow.getAs("value");
                    
                    // 更新状态：加入新值，超过长度则移除最早的值
                    recentValues.add(value);
                    if (recentValues.size() > 3) {
                        recentValues.remove(0);
                    }
                    
                    // 计算平均值
                    double avg = recentValues.stream().mapToDouble(Double::doubleValue).average().orElse(0d);
                    // 组装输出
                    outputList.add(RowFactory.create(
                        deviceId,
                        value,
                        currentRow.getAs("timestamp"),
                        avg
                    ));
                }

                // 回写状态并设置超时
                currentState.setRecentValues(recentValues);
                state.update(currentState);
                state.setTimeoutTimestamp(state.getCurrentWatermarkMs());

                return outputList.iterator();
            }
        },
        Encoders.bean(MovingAvgState.class),
        // 定义输出结构，可根据需求调整
        RowEncoder.apply(new StructType()
                .add("item", DataTypes.StringType)
                .add("current_value", DataTypes.DoubleType)
                .add("timestamp", DataTypes.TimestampType)
                .add("moving_avg_3", DataTypes.DoubleType)
        ),
        OutputMode.Update()
    );

注意事项 #

• 代码中的3对应最近3条数据的窗口大小，如果你需要的是当前行+前3行共4条的窗口，把数值改成4即可。
• 水印的超时时间1 hour可以根据业务实际的设备上报最大延迟调整，只要保证迟到的数据不会超出这个时间范围即可。
• 该方案的状态由Spark框架自动持久化，支持任务故障重启后从断点恢复计算。

内容的提问来源于stack exchange，提问作者paraflou