ReducingState和AggregatingState在注册StateDescriptor时，还需要实现一个ReduceFunction或AggregationFunction。下面的代码注册ReducingStateDescriptor时实现一个YourReduceFunction，YourReduceFunction实现了ReduceFunction。我们在ReducingState中使用add(in: T)方法向状态里增加一个元素，新元素和状态中已有数据通过ReduceFunction两两聚合。AggregatingState的使用方法与之类似。
val reducingStateDescriptor = new ReducingStateDescriptor[UserBehavior]("reducing", new YourReduceFunction, classOf[UserBehavior])