ceshi

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

+package com.byit;
+
+import com.byit.test.*;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.IterativeStream;
+import org.apache.flink.streaming.api.datastream.SplitStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+public class Test1 {
+    public static void main(String[] args) throws Exception{
+        StreamExecutionEnvironment env = StreamExecutionEnvironment
+                .getExecutionEnvironment().setBufferTimeout(1);
+
+        DataStream<Tuple2<Integer, Integer>> inputStream = env.addSource(new RandomFibonacciSource());
+
+        IterativeStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> iterativeStream =
+                inputStream.map(new TupleTransformMapFunction()).iterate(5000);
+
+        DataStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> fibonacciStream =
+                iterativeStream.map(new FibonacciCalcStepFunction());
+
+        SplitStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> branchedStream =
+                fibonacciStream.split(new FibonacciOverflowSelector());
+
+        iterativeStream.closeWith(branchedStream.select("ITERATE_FLAG"));
+
+        /*DataStream<Tuple3<Integer, Integer, Integer>> outputStream = branchedStream
+                .select("OUTPUT_FLAG").map(new BuildOutputTupleMapFunction());
+
+        outputStream.print();*/
+        inputStream.print();
+        env.execute("Streaming Iteration Example");
+    }
+
+}

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

 package com.byit;
 
+import com.byit.computer.common.FibonacciCalcStepFunction;
+import com.byit.computer.common.GroupCountSotFn;
+import com.byit.computer.common.MetricSchema;
+import com.byit.computer.common.TupleTransformMapFunction;
+import com.byit.computer.constant.PropertiesConstants;
 import com.byit.computer.resource.KafkaSource;
 import com.byit.computer.sink.KafkaSink;
 import com.byit.computer.util.ExecutionEnvUtil;
@@ -7,12 +12,19 @@ import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.common.functions.ReduceFunction;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple5;
 import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.datastream.*;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
 import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper;
 import org.apache.flink.util.Collector;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
@@ -23,6 +35,10 @@ import org.springframework.test.context.junit4.SpringRunner;
 
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 
 public class TestApplication {
 
@@ -31,32 +47,79 @@ public class TestApplication {
         final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);
         ExecutionEnvUtil.prepare(env,parameterTool);
 
-        DataStreamSource<String> data = KafkaSource.buildSource(env);
+        DataStream<Tuple2<String, Double>> inputStream = buildSource(env);
+
 
-        data.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
-            @Override
-            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
-                String[] splits = value.toLowerCase().split("\\W+");
 
-                for (String split : splits) {
-                    if (split.length() > 0) {
-                        out.collect(new Tuple2<>(split, 1));
+        env.execute("flink learning connectors kafka");
     }
+
+    public static void test1() throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(null);
+        ExecutionEnvUtil.prepare(env,parameterTool);
+
+        DataStreamSource<String> data = KafkaSource.buildSource(env);
+        data.flatMap(new GroupCountSotFn())
+                .keyBy(0)
+                .sum(1);
+        KafkaSink.setKafkaSink(data,parameterTool);
+        env.execute("flink learning connectors kafka");
     }
+
+    public static DataStream<Tuple2<String, Double>> buildSource(StreamExecutionEnvironment env) throws IllegalAccessException {
+        ParameterTool parameter = (ParameterTool) env.getConfig().getGlobalJobParameters();
+        String topic = parameter.getRequired(PropertiesConstants.METRICS_TOPIC);
+        Long time = parameter.getLong(PropertiesConstants.CONSUMER_FROM_TIME, 1L);
+        return buildSource(env, topic, time);
     }
-        }).groupBy(0)
 
-        .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
-            @Override
-            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
-                return new Tuple2<>(value1.f0, value1.f1 + value1.f1);
+    /**
+     * @param env
+     * @param topic
+     * @param time  订阅的时间
+     * @return
+     * @throws IllegalAccessException
+     */
+    public static DataStream<Tuple2<String, Double>> buildSource(StreamExecutionEnvironment env, String topic, Long time) throws IllegalAccessException {
+        ParameterTool parameterTool = (ParameterTool) env.getConfig().getGlobalJobParameters();
+        Properties props = buildKafkaProps(parameterTool);
+        FlinkKafkaConsumer011<Tuple2<String, Double>> consumer = new FlinkKafkaConsumer011<>(
+                topic,
+                new MetricSchema(),
+                props);
+        //重置offset到time时刻
+        if (time != 0L) {
+            Map<KafkaTopicPartition, Long> partitionOffset = buildOffsetByTime(props, parameterTool, time);
+            consumer.setStartFromSpecificOffsets(partitionOffset);
+        }
+        return env.addSource(consumer);
     }
-        })
-        .print();
 
-        KafkaSink.setKafkaSink(data,parameterTool);
-        //data.print();
-        env.execute("flink learning connectors kafka");
+    public static Properties buildKafkaProps(ParameterTool parameterTool) {
+        Properties props = parameterTool.getProperties();
+        props.put("bootstrap.servers", parameterTool.get(PropertiesConstants.KAFKA_BROKERS, "10.0.10.136:9092,10.0.10.137:9092,10.0.10.138:9092"));
+        props.put("zookeeper.connect", parameterTool.get(PropertiesConstants.KAFKA_ZOOKEEPER_CONNECT, "10.0.10.136:2181,10.0.10.137:2181,10.0.10.138:2181"));
+        props.put("group.id", parameterTool.get(PropertiesConstants.KAFKA_GROUP_ID, "test"));
+        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("auto.offset.reset", "latest");
+        return props;
+    }
+
+    private static Map<KafkaTopicPartition, Long> buildOffsetByTime(Properties props, ParameterTool parameterTool, Long time) {
+        props.setProperty("group.id", "query_time_" + time);
+        KafkaConsumer consumer = new KafkaConsumer(props);
+        List<PartitionInfo> partitionsFor = consumer.partitionsFor(parameterTool.getRequired(PropertiesConstants.METRICS_TOPIC));
+        Map<TopicPartition, Long> partitionInfoLongMap = new HashMap<>();
+        for (PartitionInfo partitionInfo : partitionsFor) {
+            partitionInfoLongMap.put(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()), time);
         }
+        Map<TopicPartition, OffsetAndTimestamp> offsetResult = consumer.offsetsForTimes(partitionInfoLongMap);
+        Map<KafkaTopicPartition, Long> partitionOffset = new HashMap<>();
+        offsetResult.forEach((key, value) -> partitionOffset.put(new KafkaTopicPartition(key.topic(), key.partition()), value.offset()));
 
+        consumer.close();
+        return partitionOffset;
+    }
 }

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

+package com.byit.test;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.tuple.Tuple5;
+
+public class BuildOutputTupleMapFunction extends RichMapFunction<
+        Tuple5<Integer, Integer, Integer, Integer, Integer>,
+        Tuple3<Integer, Integer, Integer>> {
+    @Override
+    public Tuple3<Integer, Integer, Integer> map(Tuple5<Integer, Integer, Integer, Integer, Integer> inputTuple) throws Exception {
+        return new Tuple3<Integer, Integer, Integer>(
+                inputTuple.f0,
+                inputTuple.f1,
+                inputTuple.f4);
+    }
+}

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

+package com.byit.test;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.tuple.Tuple5;
+
+public class FibonacciCalcStepFunction extends RichMapFunction<Tuple5<Integer, Integer, Integer, Integer, Integer>,
+                        Tuple5<Integer, Integer, Integer, Integer, Integer>> {
+    @Override
+    public Tuple5<Integer, Integer, Integer, Integer, Integer> map(Tuple5<Integer, Integer, Integer, Integer, Integer> inputTuple) throws Exception {
+        System.out.println("Step:"+inputTuple.f0 + "," + inputTuple.f1+","+inputTuple.f2+","+inputTuple.f3+","+inputTuple.f4);
+        return new Tuple5<Integer, Integer, Integer, Integer, Integer>(
+                inputTuple.f0,
+                inputTuple.f1,
+                inputTuple.f3,
+                inputTuple.f2 + inputTuple.f3,
+                ++inputTuple.f4);
+    }
+}

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

+package com.byit.test;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+
+import java.util.Collections;
+
+public class FibonacciOverflowSelector implements OutputSelector<Tuple5<Integer, Integer, Integer, Integer, Integer>> {
+    @Override
+    public Iterable<String> select(Tuple5<Integer, Integer, Integer, Integer, Integer> inputTuple) {
+            if (inputTuple.f2 < 100 && inputTuple.f3 < 100) {
+                return Collections.singleton("ITERATE_FLAG");
+            }
+
+            return Collections.singleton("OUTPUT_FLAG");
+    }
+}

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

+package com.byit.test;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+
+import java.util.Random;
+
+public class RandomFibonacciSource implements SourceFunction<Tuple2<Integer, Integer>> {
+    private Random random = new Random();
+    private volatile boolean isRunning = true;
+    private int counter = 0;
+    private int MAX_RANDOM_VALUE = 100;
+    @Override
+    public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
+        while (isRunning && counter < MAX_RANDOM_VALUE) {
+            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);
+            ctx.collect(new Tuple2<Integer, Integer>(first, second));
+            counter++;
+            Thread.sleep(50);
+        }
+    }
+
+    @Override
+    public void cancel() {
+        isRunning = false;
+    }
+}

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

+package com.byit.test;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple5;
+
+public class TupleTransformMapFunction extends RichMapFunction<Tuple2<Integer,
+        Integer>, Tuple5<Integer, Integer, Integer, Integer, Integer>> {
+    @Override
+    public Tuple5<Integer, Integer, Integer, Integer, Integer> map(Tuple2<Integer, Integer> inputTuples) throws Exception {
+        System.out.println("2->5:"+inputTuples.f0 + "," + inputTuples.f1);
+        return new Tuple5<Integer, Integer, Integer, Integer, Integer>(
+                inputTuples.f0,
+                inputTuples.f1,
+                inputTuples.f0,
+                inputTuples.f1,
+                0);
+    }
+}