flink怎么获取历史数据和当前数据进行合并计算

在阿里云Flink中，可以通过使用状态（State）来获取历史数据并进行合并计算。具体步骤如下：

1. 定义状态：在Flink程序中定义一个或多个状态变量，用于保存历史数据的计算结果。例如，可以使用ValueState、ListState等状态类型来保存历史数据。

2. 更新状态：在每次处理新数据时，将新数据与历史数据进行合并，并更新状态变量的值。例如，可以使用ValueState.update()方法或ListState.add()方法将新数据添加到历史数据中。

3. 计算结果：在每次处理完一批数据后，可以从状态变量中获取历史数据，并与当前数据进行合并计算。例如，可以使用ValueState.value()方法或ListState.get()方法获取历史数据，并使用业务逻辑对历史数据和当前数据进行合并计算，得到最终的计算结果。

在使用状态进行历史数据的合并计算时，需要考虑状态的存储和管理方式，以确保程序的性能和正确性。例如，可以使用Flink的Checkpoint机制来保证状态的一致性和可恢复性，或者使用Flink的状态后端来管理状态的存储和访问。

要获取历史数据和当前数据合并计算，可以使用 Flink 的 Stateful Functions API，它提供了分布式状态处理和事件驱动程序模型。它的一个主要好处是，它提供了一个简化且可扩展的状态处理 API，可以在大规模数据流和批处理应用中使用。

使用 Stateful Functions API 能够将数据流和批处理程序统一处理，使得历史数据和当前数据的处理变得非常容易。在处理过程中，我们可以将历史数据保存到状态中，并在处理当前数据时，从状态中获取历史数据进行计算。以下是一个演示代码的示例：

public class MergeDataStreamAndHistoricalDataStatefulFunction extends StatefulFunction {

  /**
   * 处理当前数据流的函数
   * 
   * @param input 不同类型的输入参数
   * @param context 当前函数上下文
   */
  @Override
  public void invoke(Context context, Object input) {
    if (input instanceof CurrentDataStream) {
      CurrentDataStream currentDataStream = (CurrentDataStream) input;

      // 从状态中获取历史数据
      HistoricalDataState historicalDataState = context.getState("historicalDataState", HistoricalDataState.class);
      List<HistoricalDataPoint> historicalData = historicalDataState.getHistoricalData();

      // 合并历史数据和当前数据进行计算
      List<MergedDataPoint> mergedData = mergeData(historicalData, currentDataStream.getDataStream());

      // 处理输出
      context.sendAfter(Duration.ZERO, context.call(ExternalSystemFunction.class, mergedData));
    } else if (input instanceof HistoricalDataPoint) {
      HistoricalDataPoint historicalDataPoint = (HistoricalDataPoint) input;

      // 将历史数据保存到状态中
      HistoricalDataState historicalDataState = context.getState("historicalDataState", HistoricalDataState.class);
      historicalDataState.addHistoricalDataPoint(historicalDataPoint);
    }
  }
}    

在上述代码中，我们定义了一个 MergeDataStreamAndHistoricalDataStatefulFunction 类，它包含一个 invoke() 函数。在当前数据输入时，我们从状态中获取历史数据，并将它们与当前数据合并计算。同时，我们还通过 context.call() 函数调用了一个外部系统函数，将计算后的结果发送到外部系统。在接收到历史数据时，我们将它们保存到状态中。

这只是一个简单的示例，您可以根据具体业务需求，修改和扩展 StatefulFunction 类，来进行历史数据和当前数据的合并计算。

要获取历史数据并与当前数据合并进行计算，您可以使用 Flink 的 Stateful Stream Processing 功能。具体来说，您可以在 Flink 应用程序中定义一个状态，用于存储历史数据，并将当前数据与历史数据合并以产生结果。

下面是一些示例代码，演示如何在 Flink 应用程序中使状态流计算：

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class StatefulStreamProcessing {

  public static void main(String[] args) throws Exception {
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // 数据源
    DataStream<Tuple2<String, Integer>> events = env.fromElements(
        Tuple2.of("A", 1), Tuple2.of("B", 2), Tuple2.of("A", 3), Tuple2.of("C", 4), Tuple2.of("B", 5)
    );

    // 计算
    DataStream<Tuple2<String, Integer>> result = events
        .keyBy(0)
        .flatMap(new StatefulFlatMapFunction());

    // 输出结果
    result.print();

    env.execute("Stateful Stream Processing");
  }

  public static class StatefulFlatMapFunction extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {

    private MapState<String, Integer> history;

    @Override
    public void open(Configuration config) {
      // 定义状态描述符
      MapStateDescriptor<String, Integer> historyDescriptor =
          new MapStateDescriptor<>("history", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);

      // 初始化状态
      history = getRuntimeContext().getMapState(historyDescriptor);
    }

    @Override
    public void flatMap(Tuple2<String, Integer> event, Collector<Tuple2<String, Integer>> out) throws Exception {
      // 当前数据
      String key = event.f0;
      Integer value = event.f1;

      // 历史数据
      Integer historyValue = history.get(key);
      if (historyValue == null) {
        historyValue = 0;
      }

      // 合并计算
      int sum = historyValue + value;

      // 更新状态
      history.put(key, sum);

      // 输出结果
      out.collect(Tuple2.of(key, sum));
    }
  }
}

在这个示例中，我们使用 Flink 的 MapState 存储历史数据，然后使用 RichFlatMapFunction 对当前数据进行处理，并将当前数据与历史数据合并，最后将计算结果输出。通过这种方式，您可以实现更复杂的有状态流应用。

Flink 可以通过 State 和时间窗口机制，结合 MapState 等数据结构，来实现历史数据和当前数据的合并计算。

具体实现步骤如下：

1、定义一个 ProcessFunction 类，用于获取历史数据和当前数据进行合并计算。例如：

public class MergeProcessFunction extends ProcessFunction<SensorReading, SensorData> {

    private transient MapState<String, SensorData> sensorDataState;
    private final long windowSize;

    public MergeProcessFunction(long windowSize) {
        this.windowSize = windowSize;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        MapStateDescriptor<String, SensorData> descriptor = new MapStateDescriptor<>("sensorDataState",
                String.class, SensorData.class);
        sensorDataState = getRuntimeContext().getMapState(descriptor);
    }

    @Override
    public void processElement(SensorReading sensorReading, Context context, Collector<SensorData> collector) throws Exception {
        // 从状态中获取历史数据
        SensorData sensorData = sensorDataState.get(sensorReading.id);

        // 如果历史数据为 null，则新建一个 SensorData
        if (sensorData == null) {
            sensorData = new SensorData();
            sensorData.id = sensorReading.id;
            sensorData.temperatureSum = 0.0;
            sensorData.temperatureCount = 0;
        }

        // 更新温度和温度计数
        sensorData.temperatureSum += sensorReading.temperature;
        sensorData.temperatureCount++;

        // 将更新后的 SensorData 存储到状态中
        sensorDataState.put(sensorReading.id, sensorData);

        // 注册一个定时器，用于在窗口结束时触发计算
        context.timerService().registerEventTimeTimer(context.timestamp() + windowSize);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<SensorData> out) throws Exception {
        // 计算历史数据和当前数据的平均温度
        SensorData sensorData = sensorDataState.get(ctx.getCurrentKey());
        if (sensorData != null) {
            Double averageTemperature = sensorData.temperatureSum / sensorData.temperatureCount;
            // 输出计算结果
            out.collect(new SensorData(sensorData.id, averageTemperature));
        }
        // 清空状态
        sensorDataState.clear();
    }
}

其中，SensorReading 表示传感器读数，SensorData 表示一个传感器的数据，包含传感器 ID、平均温度等。

2、在 DataStream 上应用时间窗口，然后使用 keyBy 定义键，调用 process 方法，并传递上一步定义的 ProcessFunction 对象。例如：

DataStream<SensorData> avgDataStream = stream
    .keyBy(sensorReading -> sensorReading.id)
    .timeWindow(Time.minutes(5))
    .process(new MergeProcessFunction(Time.minutes(5).toMilliseconds()));

在此示例中，我们使用了 5 分钟的时间窗口，为了方便起见，我们使用传感器 ID 作为键。然后，我们调用 process 方法，并传递 MergeProcessFunction 对象。

3、将计算结果输出到外部系统。例如：

avgDataStream.print();

在此示例中，我们简单地将计算结果输出到控制台。

值得注意的是，Flink 的窗口机制会将窗口的所有事件数据保存在 Flink 的缓存中，因此，当时间窗口较大时，可能会导致 Flink 的状态和缓存占用过多内存。为了避免此问题，您可以使用增量聚合函数，如 aggregate 方法，来限制内存使用。

在 Flink 中，可以使用 DataStream API 和 Table API/SQL 来获取历史数据和当前数据进行合并计算。

使用 DataStream API：

通过使用 Flink 的状态管理功能，可以在 DataStream 中存储历史数据。下面是一个简单的示例：

DataStream<Tuple2<String, Integer>> stream = env.fromElements(
      new Tuple2<>("foo", 1),
      new Tuple2<>("foo", 2),
      new Tuple2<>("bar", 3));

stream
      .keyBy(0)
      .flatMap(new RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {

          private ValueState<Integer> sumState;

          @Override
          public void open(Configuration config) {
              ValueStateDescriptor<Integer> descriptor =
                      new ValueStateDescriptor<Integer>("sum", Integer.class);
              sumState = getRuntimeContext().getState(descriptor);
          }

          @Override
          public void flatMap(Tuple2<String, Integer> input, Collector<Tuple2<String, Integer>> out) throws Exception {
              Integer sum = sumState.value();
              if(sum == null) {
                  sum = 0;
              }
              sum += input.f1;
              sumState.update(sum);
              out.collect(new Tuple2<>(input.f0, sum));
          }
      })
      .print();

在上面的示例中，使用 ValueState 存储了历史数据的和，然后通过 flatMap 函数计算新的和并输出。

使用 Table API/SQL：

在 Table API/SQL 中，可以使用 Flink 的时间属性和窗口操作来获取历史数据和当前数据进行合并计算。下面是一个简单的示例：

DataStream<Tuple2<String, Integer>> stream = env.fromElements(
      new Tuple2<>("foo", 1),
      new Tuple2<>("foo", 2),
      new Tuple2<>("bar", 3));

Table table = tableEnv.fromDataStream(stream, $("key"), $("value"), $("timestamp").rowtime());

table
      .window(Tumble.over(lit(10).seconds()).on($("timestamp")).as("w"))
      .groupBy($("key"), $("w"))
      .select($("key"), $("w").end(), $("value").sum().as("sum"))
      .execute()
      .print();

在上面的示例中，使用 rowtime 属性来表示时间，并通过 window 操作对数据进行分组和聚合计算。以 Tumble 为例，它会滑动一个窗口进行计算。其中，lit(10).seconds() 表示窗口大小为 10 秒。

参考链接：

• https://ci.apache.org/projects/flink/flink-docs-release-1.13/dev/stream/state/state.html
• https://ci.apache.org/projects/flink/flink-docs-release-1.13/dev/table/streaming/time_attributes.html

在Flink中，可以使用KeyedStream reduce()或KeyedStreamFold()方法来将历史数据和当前数据进行合并计算。

reduce()方法接受一个ReduceFunction参数，用于将两个相同 key 的数据进行合并计算。例如，对相同key的数据进行求和操作：

DataStream<Tuple2<String, Integer>> dataStream = ...; dataStream.keyBy(0) .reduce(new ReduceFunction<Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception { return Tuple2.of(value1.f0, value1.f1 + value2.f1); } });

fold()方法接受一个初始值和一个FoldFunction参数，用于将历史数据、当前数据和初始值进行合并计算。例如，以下代码将对相同key的数据进行求和操作，并将初始值设置为0：

DataStream<Tuple2<String, Integer>> dataStream = ...; dataStream.keyBy(0) .fold(0, newFoldFunction<Tuple2<String, Integer>, Integer>() { @Override public IntegerFold(Integer accumulator, Tuple2<String, Integer> value) throws Exception { return accumulator + value.f1; } });

reduce()和fold()方法都是针对KeyedStream的操作，因此需要先使用keyBy()方法将数据按照 key 进行分组。另外，如果您需要对历史数据和当前数据进行更复杂的合并计算，可以使用KeyedStream process()方法，自定义ProcessFunction来实现。

在Flink中，可以使用状态编程来实现历史数据和当前数据的合并计算。具体而言，可以使用Flink的状态后端来存储历史数据，并使用Flink的ProcessFunction来实现对当前数据的处理。

在ProcessFunction中，可以使用Flink的状态编程API来读取和更新状态。例如，可以使用ValueState来存储历史数据，并使用Timer来触发对当前数据的处理。当Timer触发时，可以读取历史数据和当前数据，并进行合并计算。

下面是一个示例代码，用于计算每个用户的总消费金额：

public class UserTotalExpense extends ProcessFunction<Event, Tuple2<String, Double>> { private transient ValueState totalExpenseState; @Override public void open(Configuration parameters) throws Exception { totalExpenseState = getRuntimeContext().getState( new ValueStateDescriptor<>("total-expense", Double.class)); } @Override public void processElement(Event event, Context context, Collector<Tuple2<String, Double>> collector) throws Exception { // 读取历史数据 Double totalExpense = totalExpenseState.value(); if (totalExpense == null) { totalExpense = 0.0; } // 计算当前数据 Double expense = event.getExpense(); totalExpense += expense; // 更新状态 totalExpenseState.update(totalExpense); // 设置Timer long timer = event.getTimestamp() + 24 * 60 * 60 * 1000; // 每天计算一次 context.timerService().registerEventTimeTimer(timer); } @Override public void onTimer(long timestamp, OnTimerContext context, Collector<Tuple2<String, Double>> collector) throws Exception { // 读取历史数据 Double totalExpense = totalExpenseState.value(); if (totalExpense == null) { totalExpense = 0.0; } // 输出结果 collector.collect(new Tuple2<>(context.getCurrentKey(), totalExpense)); // 清空状态 totalExpenseState.clear(); } } 在上述示例中，每个Event表示一个用户的消费记录，包括用户ID、消费金额和时间戳。在processElement方法中，使用ValueState来存储每个用户的总消费金额，并使用Timer来触发对当前数据的处理。在onTimer方法中，读取历史数据和当前数据，并进行合并计算。最后，输出每个用户的总消费金额，并清空状态。

Flink 可以通过在作业中集成历史数据和当前数据进行合并计算。常见的场景包括：

带有历史积累的窗口计算。例如，将过去一段时间内的数据进行统计分析，可以使用滑动窗口或者滚动窗口来维护历史数据，并在新数据进入时进行合并计算。

实时计算中加入维度表。例如，在实时计算中需要加入商品信息，可以使用 Flink 的 Broadcast State 或者 Table Join 等机制将维度表与实时数据进行关联，并在处理实时数据时合并计算维度信息。

要获取历史数据和当前数据进行合并计算，可以使用 Flink 的状态编程功能。具体地，可以使用 Flink 的状态来存储历史数据，并在处理新数据时将历史数据和当前数据合并计算。下面是一个简单的示例：

1. 首先，定义一个状态来存储历史数据。例如，可以使用 ValueState 来存储一个累加器的值：

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
public class MergeFunction extends RichFlatMapFunction<DataPoint, Result> {
    private transient ValueState<Double> sumState;
    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Double> descriptor = new ValueStateDescriptor<>("sumState", Double.class);
        sumState = getRuntimeContext().getState(descriptor);
    }
    @Override
    public void flatMap(DataPoint data, Collector<Result> out) throws Exception {
        Double sum = sumState.value();
        if (sum == null) {
            sum = 0.0;
        }
        sum += data.getValue();
        sumState.update(sum);
        out.collect(new Result(data.getTimestamp(), sum));
    }
}

在上述示例中，使用 ValueStateDescriptor 来创建一个 ValueState，然后在 open 方法中将其初始化。在 flatMap 方法中，首先获取历史数据的值，如果历史数据为空则初始化为 0.0，然后将历史数据和新数据进行合并计算，并更新状态的值。最后，使用 Collector 来输出结果。

1. 接下来，将该函数应用到数据流中。例如，可以使用以下代码来创建一个数据流并应用 MergeFunction：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class MergeExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<DataPoint> dataStream = env.addSource(new DataPointSource());
        DataStream<Result> resultStream = dataStream.flatMap(new MergeFunction());
        resultStream.print();
        env.execute("Merge Example");
    }
}

在上述示例中，使用 DataPointSource 来生成数据流，然后使用 MergeFunction 来对数据流进行合并计算，最后使用 print 方法来输出结果。

在实际情况中，可以根据需要修改 MergeFunction 的实现，例如，可以使用 ListState 来存储历史数据，或者使用 MapState 来存储更复杂的状态等等。

要获取历史数据和当前数据进行合并计算，需要使用Flink的状态编程功能。

首先，在Flink中定义一个状态，用于保存历史数据。可以使用Flink提供的状态后端，如MemoryStateBackend或RocksDBStateBackend，来保存状态。

然后，在Flink的DataStream中，使用KeyedStream进行分组操作，并使用ProcessFunction来处理历史数据和当前数据。ProcessFunction可以访问当前数据，还可以访问保存在状态中的历史数据，从而完成历史数据和当前数据的合并计算。

下面是一个简单的示例代码：

Copy code

DataStream<Tuple2<String, Integer>> dataStream = ...; // 输入数据流

DataStream<Tuple2<String, Integer>> resultStream = dataStream
  .keyBy(0) // 根据第一个字段进行分组
  .process(new ProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
    private ValueState<Integer> historyState;

    @Override
    public void open(Configuration parameters) throws Exception {
      // 初始化历史状态
      ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("history", Integer.class);
      historyState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
      Integer historyValue = historyState.value(); // 获取历史状态

      if (historyValue == null) {
        // 如果历史状态不存在，则将当前数据作为历史数据保存
        historyState.update(value.f1);
        out.collect(value);
      } else {
        // 如果历史状态存在，则将历史数据和当前数据进行合并计算
        int newValue = historyValue + value.f1;
        historyState.update(newValue); // 更新历史状态
        out.collect(new Tuple2<>(value.f0, newValue)); // 输出合并结果
      }
    }
  });

在这个示例代码中，我们定义了一个ValueState，用于保存历史数据。在ProcessFunction的processElement方法中，我们获取历史状态并进行处理，如果历史状态不存在，则将当前数据作为历史数据保存；如果历史状态存在，则将历史数据和当前数据进行合并计算，并输出合并结果。

需要注意的是，这个示例代码只处理了一种情况，即历史数据只有一条的情况。如果历史数据有多条，需要在状态中保存一个列表，将所有历史数据保存在其中。另外，如果历史数据和当前数据的时间戳不同，还需要使用Watermark来控制数据的处理顺序。

可以使用State（状态）机制来存储历史数据，并将历史数据和新数据进行合并计算。State是一种在Flink中存储状态信息的机制，可以在不同时间间隔的数据处理过程中保持状态信息的一致性。通常在KeyedStream中使用State来存储状态信息。

在Flink中，State主要分为三类：

1、Keyed State：基于KeyedStream的状态，即将状态绑定到特定的Key上，每个Key只能访问自己的状态信息。

2、Operator State：在操作符中共享的状态，适用于每个并行的子任务之间共享一些状态信息。

3、Broadcast State：将状态广播到所有的并行任务中。

具体实现的方式，可以根据具体的场景选择合适的状态类型，然后在对应的KeyedStream上使用State机制进行历史数据的存储和计算。常见的方式是使用ProcessFunction对数据进行处理，在ProcessFunction中使用State机制来存储和更新历史数据。

如果您需要将历史数据和当前数据进行合并计算，可以使用 Flink 的状态管理机制来实现。具体来说，在 Flink 应用程序中，可以使用 ListState 或 MapState 来存储历史数据，并通过 ListState.add() 方法或 MapState.put() 方法将当前数据添加到状态中。

然后，在处理新的输入数据时，可以通过 ListState.get() 方法或 MapState.get() 方法来获取历史数据，并将其与当前数据进行合并计算。

例如，以下代码演示了如何使用 ListState 存储历史数据，并将当前数据与历史数据进行求和：

public class MyFunction extends RichFlatMapFunction<Integer, Integer> {

private transient ListState<Integer> listState;

@Override
public void open(Configuration config) throws Exception {
    // 创建 ListState 对象
    ListStateDescriptor<Integer> descriptor = new ListStateDescriptor<>("history-data", Integer.class);
    listState = getRuntimeContext().getListState(descriptor);
}

@Override
public void flatMap(Integer input, Collector<Integer> collector) throws Exception {
    // 读取历史数据
    Iterable<Integer> historyData = listState.get();

    // 将当前数据与历史数据相加
    int sum = input;
    for (Integer data : historyData) {
        sum += data;
    }

    // 输出结果
    collector.collect(sum);

    // 将当前数据添加到历史数据中
    listState.add(input);
}

} 在上面的代码中，我们创建了一个名为 history-data 的 ListState 对象，并在 flatMap() 方法中通过 listState.get() 方法获取历史数据。然后，将当前数据与历史数据相加，输出结果，并通过 listState.add() 方法将当前数据添加到历史数据中。

需要注意的是，使用状态管理机制时需要考虑到数据倾斜和性能等问题，需要根据具体场景进行调整和优化

可以使用 Flink 中的 State 状态管理机制，将历史数据存储在状态中，与来自实时流的数据进行合并计算。

以下是一个示例：

public static class MyMapper extends RichMapFunction<Event, Result> {
        
    // 声明一个 ValueState，用于存储历史数据
    private transient ValueState<Double> history;

    public void open(Configuration parameters) {
        // 获取当前 Operator 的状态对象
        history = getRuntimeContext().getState(
                new ValueStateDescriptor<Double>("history", Double.class)
        );
    }

    public Result map(Event event) throws Exception {
        Result result = new Result();
        
        // 获取历史数据
        Double historyValue = history.value();
        if (historyValue == null) {
            historyValue = 0.0;
        }
        
        // 计算并更新结果
        Double current = event.getValue();
        Double sum = historyValue + current;
        result.setSum(sum);
        result.setCount(event.getCount() + 1);
        result.setAverage(sum / result.getCount());
        
        // 存储历史数据
        history.update(sum);

        return result;
    }
}

在上述代码中，我们通过 ValueState 存储了历史结果，并在每次处理实时事件时将历史数据和当前数据合并计算，并将新的结果保存在输出结果中。具体实现中，我们使用了 Flink 提供的 ValueStateDescriptor 来声明状态对象，并在 open 方法中通过 getRuntimeContext().getState 方法获取当前 Operator 的状态对象。在 map 方法中，我们通过 history.value() 获取历史数据，并通过 history.update(sum) 方法将新的计算结果存储到状态中。