结构更改

byit-screen-computer/src/test/java/com/byit/Test1.java

@@ -17,21 +17,20 @@ public class Test1 {
         DataStream<Tuple2<Integer, Integer>> inputStream = env.addSource(new RandomFibonacciSource());
 
         IterativeStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> iterativeStream =
-                inputStream.map(new TupleTransformMapFunction()).iterate(5000);
+                inputStream.map(new TupleTransformMapFunction()).iterate(50000);//2->5 last:0
 
         DataStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> fibonacciStream =
-                iterativeStream.map(new FibonacciCalcStepFunction());
+                iterativeStream.map(new FibonacciCalcStepFunction());//斐波那契额Step,算法
 
         SplitStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> branchedStream =
-                fibonacciStream.split(new FibonacciOverflowSelector());
+                fibonacciStream.split(new FibonacciOverflowSelector());//限制条件
 
-        iterativeStream.closeWith(branchedStream.select("ITERATE_FLAG"));
+        iterativeStream.closeWith(branchedStream.select("ITERATE_FLAG"));//下次递归数据total
 
-        /*DataStream<Tuple3<Integer, Integer, Integer>> outputStream = branchedStream
-                .select("OUTPUT_FLAG").map(new BuildOutputTupleMapFunction());
+        DataStream<Tuple3<Integer, Integer, Integer>> outputStream = branchedStream
+                .select("OUTPUT_FLAG").map(new BuildOutputTupleMapFunction());//本次输出
 
-        outputStream.print();*/
-        inputStream.print();
+        outputStream.print();
         env.execute("Streaming Iteration Example");
     }
 

byit-screen-computer/src/test/java/com/byit/test/BuildOutputTupleMapFunction.java

@@ -9,6 +9,7 @@ public class BuildOutputTupleMapFunction extends RichMapFunction<
         Tuple3<Integer, Integer, Integer>> {
     @Override
     public Tuple3<Integer, Integer, Integer> map(Tuple5<Integer, Integer, Integer, Integer, Integer> inputTuple) throws Exception {
+        System.out.println("output:"+inputTuple.f0+","+inputTuple.f1+","+inputTuple.f4);
         return new Tuple3<Integer, Integer, Integer>(
                 inputTuple.f0,
                 inputTuple.f1,

byit-screen-computer/src/test/java/com/byit/test/RandomFibonacciSource.java

@@ -16,8 +16,7 @@ public class RandomFibonacciSource implements SourceFunction<Tuple2<Integer, Int
             int first = random.nextInt(MAX_RANDOM_VALUE / 2 - 1) + 1;
             int second = random.nextInt(MAX_RANDOM_VALUE / 2 -1) + 1;
 
-            if (first > second) continue;
-            System.out.println("0->2:"+first+","+second);
+            System.out.println("s:"+counter+":"+first+","+second);
             ctx.collect(new Tuple2<Integer, Integer>(first, second));
             counter++;
             Thread.sleep(50);