1

src/main/java/org/myorg/quickstart/BatchJob.java

@@ -1,4 +1,4 @@
-/*
+package org.myorg.quickstart;/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -16,8 +16,6 @@
  * limitations under the License.
  */
 
-package org.myorg.quickstart;
-
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
@@ -29,7 +27,6 @@ public class BatchJob {
 
 	public static void main(String[] args) throws Exception {
 
-		// set up the execution environment
 		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		env.setParallelism(5);
 		// get input data

src/main/java/org/myorg/quickstart/State09/BatchJob1.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.myorg.quickstart.State09;
+
+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.api.common.typeinfo.TypeHint;
+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.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.Collector;
+
+public class BatchJob1 {
+
+
+	public static void main(String[] args) throws Exception {
+
+		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+		env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 5L), Tuple2.of(1L, 2L))
+				.keyBy(0)
+				.flatMap(new CountWindowAverage())
+				.printToErr();
+
+        env.execute("submit job");
+
+	}
+
+
+    public static class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
+
+        private transient ValueState<Tuple2<Long, Long>> sum;
+        public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
+
+            // 访问ValueState
+            Tuple2<Long, Long> currentSum = sum.value();
+
+            // 更新
+            currentSum.f0 += 1;
+
+            // 第二个元素加1
+            currentSum.f1 += input.f1;
+
+            // 更新state
+            sum.update(currentSum);
+
+            // 如果count的值大于等于2，求知道并清空state
+            if (currentSum.f0 >= 2) {
+                out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
+                sum.clear();
+            }
+    }
+
+
+    public void open(Configuration config) {
+        ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
+                new ValueStateDescriptor<>(
+                        "average", // state的名字
+                        TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}),
+                        Tuple2.of(0L, 0L)); // 设置默认值
+        sum = getRuntimeContext().getState(descriptor);
+    }
+}
+}