{"id":153447,"date":"2020-09-10T20:10:08","date_gmt":"2020-09-10T12:10:08","guid":{"rendered":"http:\/\/4563.org\/?p=153447"},"modified":"2020-09-10T20:10:08","modified_gmt":"2020-09-10T12:10:08","slug":"%e6%b5%85%e8%b0%88flink-state%e7%8a%b6%e6%80%81%e7%ae%a1%e7%90%86%e6%9c%ba%e5%88%b6","status":"publish","type":"post","link":"http:\/\/4563.org\/?p=153447","title":{"rendered":"\u6d45\u8c08flink state\u72b6\u6001\u7ba1\u7406\u673a\u5236"},"content":{"rendered":"

<\/a>state \u7b80\u8ff0<\/h2>\n

Apache Flink\u00ae \u2014 Stateful Computations over Data Streams<\/strong>,flink\u662f\u4e00\u4e2a\u9ed8\u8ba4\u5c31\u6709\u72b6\u6001\u7684\u5206\u5206\u6790\u5f15\u64ce\uff0c\u9488\u5bf9\u6d41\u5931\u8ba1\u7b97\u5f15\u64ce\u4e2d\u7684\u6570\u636e\u5f80\u5f80\u662f\u8f6c\u77ac\u5373\u901d\uff0c\u4f46\u5728flink\u771f\u5b9e\u4e1a\u52a1\u573a\u666f\u786e\u4e0d\u80fd\u8fd9\u6837\uff0c\u4ec0\u4e48\u90fd\u4e0d\u80fd\u7559\u4e0b\uff0c\u80af\u5b9a\u662f\u9700\u8981\u6709\u6570\u636e\u7559\u4e0b\u7684\uff0c\u9488\u5bf9\u8fd9\u4e9b\u6570\u636e\u7559\u4e0b\u6765\u5b58\u50a8\u4e0b\u6765\uff0c\u5728flink\u4e2d\u53eb\u505astate\uff0c\u4e2d\u6587\u53ef\u4ee5\u7ffb\u8bd1\u6210\u72b6\u6001\u3002<\/p>\n

<\/a>state \u7c7b\u578b<\/h2>\n

Flink\u4e2d\u6709\u4e24\u79cd\u57fa\u672c\u7c7b\u578b\u7684State, \u5206\u522b\u4e3a Keyed State(\u952e\u63a7\u72b6\u6001) \u548c Operator State(\u7b97\u5b50\u72b6\u6001)\u3002<\/p>\n

<\/a>keyed State(\u952e\u63a7\u72b6\u6001)<\/h2>\n

Keyed State\uff1a\u987e\u540d\u601d\u4e49\u5c31\u662f\u57fa\u4e8e KeyedStream \u4e0a\u7684\u72b6\u6001\uff0c\u8fd9\u4e2a\u72b6\u6001\u662f\u8ddf\u7279\u5b9a\u7684Key\u7ed1\u5b9a\u7684\u3002KeyedStrean \u6d41\u4e0a\u7684\u6bcf\u4e00\u4e2aKey\uff0c\u90fd\u5bf9\u5e94\u4e00\u4e2a State\u3002Flink\u9488\u5bf9Keyed State\u63d0\u4f9b\u4e86\u4ee5\u4e0b\u516d\u79cd\u7c7b\u53ef\u4ee5\u4fdd\u5b58 State \u7684\u6570\u636e\u7ed3\u6784\u7c7b\u578b\u3002<\/p>\n

<\/a>keyed state\u6258\u7ba1\u72b6\u6001\u6709\u516d\u79cd\u7c7b\u578b<\/h3>\n

<\/a>ValueState<\/h4>\n

\u4fdd\u5b58\u4e00\u4e2a\u53ef\u4ee5\u66f4\u65b0\u548c\u68c0\u7d22\u7684\u503c\uff08\u5982\u4e0a\u6240\u8ff0\uff0c\u6bcf\u4e2a\u503c\u90fd\u5bf9\u5e94\u5230\u5f53\u524d\u7684\u8f93\u5165\u6570\u636e\u7684 Key\uff0c\u56e0\u6b64\u7b97\u5b50\u63a5\u6536\u5230\u7684\u6bcf\u4e2aKey\u90fd\u53ef\u80fd\u5bf9\u5e94\u4e00\u4e2a\u503c\uff09\u3002\u8fd9\u4e2a\u503c\u53ef\u4ee5\u901a\u8fc7 update(T)\u8fdb\u884c\u66f4\u65b0\uff0c\u901a\u8fc7 T value() \u8fdb\u884c\u68c0\u7d22\u3002<\/p>\n

<\/a>ListState<\/h4>\n

\u4fdd\u5b58\u4e00\u4e2a\u5143\u7d20\u7684\u5217\u8868\u3002\u53ef\u4ee5\u5f80\u8fd9\u4e2a\u5217\u8868\u4e2d\u8ffd\u52a0\u6570\u636e\uff0c\u5e76\u5728\u5f53\u524d\u7684\u5217\u8868\u4e0a\u8fdb\u884c\u68c0\u7d22\u3002\u53ef\u4ee5\u901a\u8fc7 add(T) \u6216\u8005 addAll(List) \u8fdb\u884c\u6dfb\u52a0\u5143\u7d20\uff0c\u901a\u8fc7 Iterable get() \u83b7\u53d6\u6574\u4e2a\u5217\u8868\u3002\u8fd8\u53ef\u4ee5\u901a\u8fc7 update(List) \u8986\u76d6\u5f53\u524d\u7684\u5217\u8868\u3002<\/p>\n

<\/a>MapState<UK,UV><\/h4>\n

\u7ef4\u62a4\u4e86\u4e00\u4e2a\u6dfb\u52a0\u6620\u5c04\u5217\u8868\u3002\u4f60\u53ef\u4ee5\u6dfb\u52a0\u952e\u503c\u5bf9\u5230\u72b6\u6001\u4e2d\uff0c\u4e5f\u53ef\u4ee5\u83b7\u5f97\u53cd\u6620\u5f53\u524d\u6240\u6709\u6620\u5c04\u7684\u8fed\u4ee3\u5668\u3002\u4f7f\u7528 put(UK,UV) \u6216\u8005 putAll(Map<UK,UV>) \u5206\u522b\u68c0\u7d22\u6620\u5c04\u3001\u952e\u548c\u503c\u7684\u53ef\u8fed\u4ee3\u89c6\u56fe\u3002<\/p>\n

<\/a>ReducingState<\/h4>\n

\u4fdd\u5b58\u4e00\u4e2a\u5355\u503c\uff0c\u8868\u793a\u6dfb\u52a0\u5230\u72b6\u6001\u7684\u6240\u6709\u805a\u5408\u3002\u63a5\u53e3\u4e0e ListState \u7c7b\u4f3c\uff0c\u4f7f\u7528 add(T)\u589e\u52a0\u5143\u7d20\uff0c\u4f1a\u4f7f\u7528\u63d0\u4f9b\u7684 ReduceFunction \u8fdb\u884c\u805a\u5408<\/p>\n

<\/a>AggregatingState<IN.OUT><\/h4>\n

\u4fdd\u7559\u4e00\u4e2a\u5355\u503c\uff0c\u8868\u793a\u6dfb\u52a0\u5230\u72b6\u6001\u7684\u6240\u4ee5\u503c\u7684\u805a\u5408\u3002\u548cReducingState \u76f8\u53cd\u7684\u989d\u662f\u805a\u5408\u7c7b\u578b\u53ef\u80fd\u4e0e\u6dfb\u52a0\u5230\u72b6\u6001\u7684\u5143\u7d20\u7684\u7c7b\u578b\u4e0d\u540c\u3002\u63a5\u53e3\u4e0e ListState \u7c7b\u4f3c\uff0c\u4f46\u4f7f\u7528 add(IN) \u6dfb\u52a0\u7684\u5143\u7d20\u4f1a\u7528\u6307\u5b9a\u7684 AggregateFunction \u8fdb\u884c\u805a\u5408\u3002<\/p>\n

<\/a>FoldingState<T,ACC><\/h4>\n

\u4fdd\u7559\u4e00\u4e2a\u5355\u503c\uff0c\u8868\u793a\u6dfb\u52a0\u5230\u72b6\u6001\u7684\u6240\u6709\u503c\u7684\u805a\u5408\u3002\u4e0e ReducingState \u76f8\u53cd\uff0c\u805a\u5408\u7c7b\u578b\u53ef\u80fd\u4e0e\u6dfb\u52a0\u5230\u72b6\u6001\u7684\u5143\u7d20\u7c7b\u578b\u4e0d\u540c\u3002\u63a5\u53e3\u4e0eListState\u7c7b\u578b\uff0c\u4f46\u4f7f\u7528 add(T) \u6dfb\u52a0\u7684\u5143\u7d20\u4f1a\u7528\u6307\u5b9a\u7684FoldFunction \u6298\u53e0\u6210\u805a\u5408\u503c\u3002<\/p>\n

<\/a>Keyed State\u6848\u4f8b\u6f14\u793a<\/h3>\n

<\/a>ValueState \u4f7f\u7528valueState\u5b9e\u73b0\u5e73\u5747\u503c\u6c42\u53d6<\/h3>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
23<\/span>
24<\/span>
25<\/span>
26<\/span>
27<\/span>
28<\/span>
29<\/span>
30<\/span>
31<\/span>
32<\/span>
33<\/span>
34<\/span>
35<\/span>
36<\/span>
37<\/span>
38<\/span>
39<\/span>
40<\/span>
41<\/span>
42<\/span>
43<\/span>
44<\/span>
45<\/span>
46<\/span>
47<\/span>
48<\/span>
49<\/span>
50<\/span>
51<\/span>
52<\/span>
53<\/span>
54<\/span>
55<\/span>
56<\/span>
57<\/span>
58<\/span>
59<\/span>
60<\/span>
<\/pre>\n<\/td>\n
\n
package<\/span> xin.studytime.scala<\/span>
<\/span>
import<\/span> org.apache.flink.api.common.functions.RichFlatMapFunction<\/span><\/span>
import<\/span> org.apache.flink.api.common.state.{ValueState<\/span>, ValueStateDescriptor<\/span>}<\/span>
import<\/span> org.apache.flink.configuration.Configuration<\/span><\/span>
import<\/span> org.apache.flink.streaming.api.scala.StreamExecutionEnvironment<\/span><\/span>
import<\/span> org.apache.flink.util.Collector<\/span><\/span>
<\/span>
object<\/span> ValueStateOperate<\/span> <\/span>{<\/span>
  def<\/span> main<\/span><\/span>(args: Array<\/span>[String<\/span>]): Unit<\/span> = {<\/span>
    val<\/span> env = StreamExecutionEnvironment<\/span>.getExecutionEnvironment<\/span>
    import<\/span> org.apache.flink.api.scala._<\/span>
<\/span>
    env.fromCollection(List<\/span>(<\/span>
      (1<\/span>L, 3<\/span>d),<\/span>
      (1<\/span>L, 5<\/span>d),<\/span>
      (1<\/span>L, 7<\/span>d),<\/span>
      (1<\/span>L, 4<\/span>d),<\/span>
      (1<\/span>L, 2<\/span>d)<\/span>
    ))<\/span>
      .keyBy(_._1)<\/span>
      .flatMap(new<\/span> CountAverageWithValue<\/span>())<\/span>
      .print()<\/span>
    env.execute()<\/span>
  }<\/span>
}<\/span>
<\/span>
class<\/span> CountAverageWithValue<\/span> extends<\/span> RichFlatMapFunction<\/span>[(Long<\/span>, Double<\/span>), (Long<\/span>, Double<\/span>)] <\/span>{<\/span>
  \/\/\u5b9a\u4e49ValueState\u7c7b\u578b\u7684\u53d8\u91cf<\/span><\/span>
<\/span>
  private<\/span> var<\/span> sum: ValueState<\/span>[(Long<\/span>,Double<\/span>)] = _<\/span>
  override<\/span> def<\/span> open<\/span><\/span>(parameters: Configuration<\/span>): Unit<\/span> = {<\/span>
    \/\/\u521d\u59cb\u5316\u83b7\u53d6\u5386\u53f2\u72b6\u6001\u7684\u503c<\/span><\/span>
    val<\/span> average = new<\/span> ValueStateDescriptor<\/span>[(Long<\/span>, Double<\/span>)](\"average\"<\/span>, classOf[(Long<\/span>, Double<\/span>)])<\/span>
    sum = getRuntimeContext.getState(average)<\/span>
  }<\/span>
<\/span>
  override<\/span> def<\/span> flatMap<\/span><\/span>(input: (Long<\/span>, Double<\/span>), out: Collector<\/span>[(Long<\/span>, Double<\/span>)]): Unit<\/span> = {<\/span>
    \/\/ access the state value<\/span><\/span>
    val<\/span> tmpCurrentSum = sum.value<\/span>
    \/\/ If it hasn't been used before, it will be null<\/span><\/span>
    val<\/span> currentSum = if<\/span> (tmpCurrentSum != null<\/span>) {<\/span>
      tmpCurrentSum<\/span>
    } else<\/span> {<\/span>
      (0<\/span>L, 0<\/span>d)<\/span>
    }<\/span>
    \/\/ update the count<\/span><\/span>
    val<\/span> newSum = (currentSum._1 + 1<\/span>, currentSum._2 + input._2)<\/span>
<\/span>
    \/\/ update the state<\/span><\/span>
    sum.update(newSum)<\/span>
<\/span>
    \/\/ if the count reaches 2, emit the average and clear the state<\/span><\/span>
    if<\/span> (newSum._1 >= 2<\/span>) {<\/span>
      out.collect((input._1, newSum._2 \/ newSum._1))<\/span>
      \/\/\u5c06\u72b6\u6001\u6e05\u9664<\/span><\/span>
      \/\/sum.clear()<\/span><\/span>
    }<\/span>
  }<\/span>
}<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>ListState \u6c42\u53d6\u6570\u636e\u5e73\u5747\u503c<\/h3>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
23<\/span>
24<\/span>
25<\/span>
26<\/span>
27<\/span>
28<\/span>
29<\/span>
30<\/span>
31<\/span>
32<\/span>
33<\/span>
34<\/span>
35<\/span>
36<\/span>
37<\/span>
38<\/span>
39<\/span>
40<\/span>
41<\/span>
42<\/span>
43<\/span>
44<\/span>
45<\/span>
46<\/span>
47<\/span>
48<\/span>
49<\/span>
50<\/span>
51<\/span>
52<\/span>
53<\/span>
54<\/span>
55<\/span>
56<\/span>
57<\/span>
58<\/span>
<\/pre>\n<\/td>\n
\n
package<\/span> xin.studytime.scala<\/span>
<\/span>
import<\/span> java.lang<\/span>
import<\/span> java.util.Collections<\/span><\/span>
<\/span>
import<\/span> org.apache.flink.api.common.functions.RichFlatMapFunction<\/span><\/span>
import<\/span> org.apache.flink.api.common.state.{ListState<\/span>, ListStateDescriptor<\/span>}<\/span>
import<\/span> org.apache.flink.configuration.Configuration<\/span><\/span>
import<\/span> org.apache.flink.streaming.api.scala.StreamExecutionEnvironment<\/span><\/span>
import<\/span> org.apache.flink.util.Collector<\/span><\/span>
import<\/span> scala.collection.JavaConverters<\/span>._<\/span>
<\/span>
object<\/span> ListStateOperate<\/span> <\/span>{<\/span>
  def<\/span> main<\/span><\/span>(args: Array<\/span>[String<\/span>]): Unit<\/span> = {<\/span>
    val<\/span> env = StreamExecutionEnvironment<\/span>.getExecutionEnvironment<\/span>
    import<\/span> org.apache.flink.api.scala._<\/span>
    env.fromCollection(List<\/span>(<\/span>
      (1<\/span>L, 3<\/span>d),<\/span>
      (1<\/span>L, 5<\/span>d),<\/span>
      (1<\/span>L, 7<\/span>d),<\/span>
      (2<\/span>L, 4<\/span>d),<\/span>
      (2<\/span>L, 2<\/span>d),<\/span>
      (2<\/span>L, 6<\/span>d)<\/span>
    )).keyBy(_._1)<\/span>
      .flatMap(new<\/span> CountAverageWithList<\/span>)<\/span>
      .print()<\/span>
    env.execute()<\/span>
  }<\/span>
}<\/span>
<\/span>
class<\/span> CountAverageWithList<\/span> extends<\/span> RichFlatMapFunction<\/span>[(Long<\/span>, Double<\/span>), (Long<\/span>, Double<\/span>)] <\/span>{<\/span>
  private<\/span> var<\/span> elementsByKey: ListState<\/span>[(Long<\/span>, Double<\/span>)] = _<\/span>
<\/span>
  override<\/span> def<\/span> open<\/span><\/span>(parameters: Configuration<\/span>): Unit<\/span> = {<\/span>
    val<\/span> listState = new<\/span> ListStateDescriptor<\/span>[(Long<\/span>, Double<\/span>)](\"listState\"<\/span>, classOf[(Long<\/span>, Double<\/span>)])<\/span>
    elementsByKey = getRuntimeContext.getListState(listState)<\/span>
  }<\/span>
<\/span>
  override<\/span> def<\/span> flatMap<\/span><\/span>(in: (Long<\/span>, Double<\/span>), collector: Collector<\/span>[(Long<\/span>, Double<\/span>)]): Unit<\/span> = {<\/span>
    val<\/span> currentState: lang.Iterable<\/span>[(Long<\/span>, Double<\/span>)] = elementsByKey.get()<\/span>
<\/span>
    if<\/span> (currentState == null<\/span>) {<\/span>
      elementsByKey.addAll(Collections<\/span>.emptyList())<\/span>
    }<\/span>
    elementsByKey.add(in)<\/span>
    val<\/span> allElements: Iterator<\/span>[(Long<\/span>, Double<\/span>)] = elementsByKey.get().iterator().asScala<\/span>
    val<\/span> allElementList: List<\/span>[(Long<\/span>, Double<\/span>)] = allElements.toList<\/span>
    if<\/span> (allElementList.size > 3<\/span>) {<\/span>
      var<\/span> count = 0<\/span>L<\/span>
      var<\/span> sum = 0<\/span>d<\/span>
      for<\/span> (eachElement <- allElementList) {<\/span>
        count += 1<\/span><\/span>
        sum += eachElement._2<\/span>
      }<\/span>
      collector.collect(in._1, sum \/ count)<\/span>
    }<\/span>
  }<\/span>
}<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>MapState \u6c42\u53d6\u6570\u636e\u5e73\u5747\u503c<\/h3>\n
\n\n\n
\n
1<\/span>
2<\/span>
3<\/span>
4<\/span>
5<\/span>
6<\/span>
7<\/span>
8<\/span>
9<\/span>
10<\/span>
11<\/span>
12<\/span>
13<\/span>
14<\/span>
15<\/span>
16<\/span>
17<\/span>
18<\/span>
19<\/span>
20<\/span>
21<\/span>
22<\/span>
23<\/span>
24<\/span>
25<\/span>
26<\/span>
27<\/span>
28<\/span>
29<\/span>
30<\/span>
31<\/span>
32<\/span>
33<\/span>
34<\/span>
35<\/span>
36<\/span>
37<\/span>
38<\/span>
39<\/span>
40<\/span>
41<\/span>
42<\/span>
43<\/span>
44<\/span>
45<\/span>
46<\/span>
47<\/span>
48<\/span>
49<\/span>
50<\/span>
51<\/span>
52<\/span>
53<\/span>
54<\/span>
55<\/span>
56<\/span>
57<\/span>
<\/pre>\n<\/td>\n
\n
package<\/span> xin.studytime.scala<\/span>
<\/span>
import<\/span> java.util.UUID<\/span><\/span>
<\/span>
import<\/span> org.apache.flink.api.common.functions.RichFlatMapFunction<\/span><\/span>
import<\/span> org.apache.flink.configuration.Configuration<\/span><\/span>
import<\/span> org.apache.flink.streaming.api.scala.StreamExecutionEnvironment<\/span><\/span>
import<\/span> org.apache.flink.util.Collector<\/span><\/span>
import<\/span> org.apache.flink.api.common.state.{MapState<\/span>, MapStateDescriptor<\/span>}<\/span>
<\/span>
object<\/span> MapStateOperate<\/span> <\/span>{<\/span>
  def<\/span> main<\/span><\/span>(args: Array<\/span>[String<\/span>]): Unit<\/span> = {<\/span>
    val<\/span> environment = StreamExecutionEnvironment<\/span>.getExecutionEnvironment<\/span>
    import<\/span> org.apache.flink.api.scala._<\/span>
<\/span>
    environment.fromCollection(List<\/span>(<\/span>
      (1<\/span>L, 3<\/span>d),<\/span>
      (1<\/span>L, 5<\/span>d),<\/span>
      (1<\/span>L, 7<\/span>d),<\/span>
      (2<\/span>L, 4<\/span>d),<\/span>
      (2<\/span>L, 2<\/span>d),<\/span>
      (2<\/span>L, 6<\/span>d)<\/span>
    )).keyBy(_._1).flatMap(new<\/span> CountAverageMapState<\/span>).print()<\/span>
<\/span>
    environment.execute()<\/span>
  }<\/span>
}<\/span>
<\/span>
class<\/span> CountAverageMapState<\/span> extends<\/span> RichFlatMapFunction<\/span>[(Long<\/span>, Double<\/span>), (Long<\/span>, Double<\/span>)] <\/span>{<\/span>
<\/span>
  private<\/span> var<\/span> mapState: MapState<\/span>[String<\/span>, Double<\/span>] = _<\/span>
<\/span>
  override<\/span> def<\/span> open<\/span><\/span>(parameters: Configuration<\/span>): Unit<\/span> = {<\/span>
    val<\/span> mapStateOperate = new<\/span> MapStateDescriptor<\/span>[String<\/span>, Double<\/span>](\"mapStateOperate\"<\/span>, classOf[String<\/span>], classOf[Double<\/span>])<\/span>
    mapState = getRuntimeContext.getMapState(mapStateOperate)<\/span>
  }<\/span>
<\/span>
  override<\/span> def<\/span> flatMap<\/span><\/span>(in: (Long<\/span>, Double<\/span>), out: Collector<\/span>[(Long<\/span>, Double<\/span>)]): Unit<\/span> = {<\/span>
<\/span>
    \/\/\u5c06\u76f8\u540c\u7684key\u5bf9\u5e94\u7684\u6570\u636e\u653e\u5230\u4e00\u4e2amap\u96c6\u5408\u5f53\u4e2d\u53bb\uff0c\u5c31\u662f\u8fd9\u79cd\u5bf9\u5e94  key -> Map((key1, value1),(key2, value2))<\/span><\/span>
    \/\/\u6bcf\u6b21\u90fd\u6784\u5efa\u4e00\u4e2amap\u96c6\u5408<\/span><\/span>
    mapState.put(UUID<\/span>.randomUUID().toString, in._2)<\/span>
    import<\/span> scala.collection.JavaConverters<\/span>._<\/span>
<\/span>
    \/\/\u83b7\u53d6map\u96c6\u5408\u5f53\u4e2d\u6240\u6709\u7684value\uff0c\u6211\u4eec\u6bcf\u6b21\u5c06\u6570\u636e\u7684value\u7ed9\u653e\u5230map\u7684value\u91cc\u9762\u53bb<\/span><\/span>
    val<\/span> listState: List<\/span>[Double<\/span>] = mapState.values().iterator().asScala.toList<\/span>
    if<\/span> (listState.size >= 3<\/span>) {<\/span>
      var<\/span> count = 0<\/span>L<\/span>
      var<\/span> sum = 0<\/span>d<\/span>
      for<\/span> (eachState <- listState) {<\/span>
        count += 1<\/span><\/span>
        sum += eachState<\/span>
      }<\/span>
      out.collect(in._1, sum \/ count)<\/span>
    }<\/span>
  }<\/span>
}<\/span>
<\/pre>\n<\/td>\n<\/tr>\n<\/table>\n<\/figure>\n
<\/a>ReducingState\u3001AggregatingState\u7b49\u8003\u8651\u5b9e\u9645\u573a\u666f\u4f7f\u7528\u4e0d\u591a\uff0c\u5c31\u4e0d\u5728\u6b64\u5217\u4e3e\u3002<\/h3>\n
<\/a>Operator State(\u7b97\u5b50\u72b6\u6001)<\/h2>\n
Operator State \u4e0e Key \u65e0\u5173\uff0c\u800c\u662f\u4e0e Operator \u7ed1\u5b9a\uff0c\u6574\u4e2a Operator \u53ea\u5bf9\u5e94\u4e00\u4e2a State\u3002\u6bd4\u5982\uff1aFlink \u4e2d\u7684 Kafka Connector \u5c31\u4f7f\u7528\u4e86 Operator State\uff0c\u5b83\u4f1a\u5728\u6bcf\u4e2a Connector \u5b9e\u4f8b\u4e2d\uff0c\u4fdd\u5b58\u8be5\u5b9e\u4f8b\u6d88\u8d39 Topic \u7684\u6240\u6709\uff08partition\uff0coffset\uff09\u6620\u5c04\uff0coperator state \u53ea\u6709\u4e00\u79cd\u6258\u7ba1\u72b6\u6001\uff1aValueState<\/p>\n
<\/a>Operator State\u6848\u4f8b\u6f14\u793a<\/h3>\n
<\/a>Flink\u7684\u72b6\u6001\u7ba1\u7406\u4e4bState Backend<\/h2>\n
\u9ed8\u8ba4\u60c5\u51b5\u4e0b\uff0cstate\u4f1a\u4fdd\u5b58\u5728taskmanager\u7684\u5185\u5b58\u4e2d\uff0ccheckpoint\u4f1a\u5b58\u50a8\u5728JobManager\u7684\u5185\u5b58\u4e2d\u3002state \u7684\u5b58\u50a8\u548ccheckpoint\u7684\u4f4d\u7f6e\u53d6\u51b3\u4e8eState Backend\u7684\u914d\u7f6e\u3002Flink\u4e00\u5171\u63d0\u4f9b\u4e863\u79cdStateBackend\uff0cMemoryStateBackend\u57fa\u4e8e\u5185\u5b58\u5b58\u50a8\u3001FsStateBackend\u57fa\u4e8e\u6587\u4ef6\u7cfb\u7edf\u5b58\u50a8\u3001RocksDBStateBackend\u57fa\u4e8e\u6570\u636e\u5e93\u5b58\u50a8\u3002\u53ef\u4ee5\u901a\u8fc7  StreamExecutionEnvironment.setStateBackend(...)<\/code>\u6765\u8bbe\u7f6estate\u5b58\u50a8\u7684\u4f4d\u7f6e\u3002<\/p>\n
<\/a>MemoryStateBackend<\/h3>\n
\u5c06\u6570\u636e\u6301\u4e45\u5316\u72b6\u6001\u5b58\u50a8\u5230\u5185\u5b58\u5f53\u4e2d\uff0cstate\u6570\u636e\u4fdd\u5b58\u5728java\u5806\u5185\u5b58\u4e2d\uff0c\u6267\u884ccheckpoint\u7684\u65f6\u5019\uff0c\u4f1a\u628astate\u7684\u5feb\u7167\u6570\u636e\u4fdd\u5b58\u5230jobmanager\u7684\u5185\u5b58\u4e2d\u3002\u57fa\u4e8e\u5185\u5b58\u7684state backend\u5728\u751f\u4ea7\u73af\u5883\u4e0b\u4e0d\u5efa\u8bae\u4f7f\u7528\u3002<\/p>\n
\"\u6d45\u8c08flink<\/p>\n
<\/a>\u4ee3\u7801\u914d\u7f6e<\/h4>\n
environment.setStateBackend(new MemoryStateBackend())<\/code> \u8bbe\u7f6eMemoryStateBackend\u3002<\/p>\n
<\/a>\u4f7f\u7528\u573a\u666f<\/h4>\n