结构更改

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.api.java.tuple.*;
 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.datastream.WindowedStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 
 public class Test1 {
     public static void main(String[] args) throws Exception{
@@ -17,21 +18,42 @@ 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(50000);//2->5 last:0
+                inputStream.map(new TupleTransformMapFunction()).iterate(50000);//2->5 last:0.timeWindow(Time.minutes(5))
 
         DataStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> fibonacciStream =
                 iterativeStream.map(new FibonacciCalcStepFunction());//斐波那契额Step,算法
 
+        WindowedStream<Tuple5<Integer, Integer, Integer, Integer, Integer>, Tuple, TimeWindow> windowedStream =
+                fibonacciStream.keyBy(0).timeWindow(Time.minutes(5));
+
         SplitStream<Tuple5<Integer, Integer, Integer, Integer, Integer>> branchedStream =
                 fibonacciStream.split(new FibonacciOverflowSelector());//限制条件
 
         iterativeStream.closeWith(branchedStream.select("ITERATE_FLAG"));//下次递归数据total
 
-        DataStream<Tuple3<Integer, Integer, Integer>> outputStream = branchedStream
+        DataStream<Tuple4<Integer,Integer, Integer, Integer>> outputStream = branchedStream
                 .select("OUTPUT_FLAG").map(new BuildOutputTupleMapFunction());//本次输出
 
         outputStream.print();
         env.execute("Streaming Iteration Example");
     }
 
+    public void test1() 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(50000);//2->5 last:0.timeWindow(Time.minutes(5))
+
+
+        WindowedStream<Tuple5<Integer, Integer, Integer, Integer, Integer>, Tuple, TimeWindow> windowedStream =
+                iterativeStream.keyBy(0).timeWindow(Time.minutes(5));
+
+
+
+        inputStream.print();
+        env.execute("Streaming Iteration Example");
+    }
 }

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

 package com.byit;
 
+import com.byit.common.utils.GsonUtil;
 import com.byit.computer.common.FibonacciCalcStepFunction;
 import com.byit.computer.common.GroupCountSotFn;
 import com.byit.computer.common.MetricSchema;
@@ -9,13 +10,17 @@ import com.byit.computer.resource.KafkaSource;
 import com.byit.computer.sink.KafkaSink;
 import com.byit.computer.util.ExecutionEnvUtil;
 import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.FoldFunction;
 import org.apache.flink.api.common.functions.ReduceFunction;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple;
 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.*;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.time.Time;
 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;
@@ -43,23 +48,46 @@ import java.util.Properties;
 public class TestApplication {
 
     public static void main(String[] args) throws Exception {
-        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);
-        ExecutionEnvUtil.prepare(env,parameterTool);
-
-        DataStream<Tuple2<String, Double>> inputStream = buildSource(env);
-
-
-
-        env.execute("flink learning connectors kafka");
+        test1();
     }
 
     public static void test1() throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(null);
+        final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(new String[]{});
         ExecutionEnvUtil.prepare(env,parameterTool);
 
         DataStreamSource<String> data = KafkaSource.buildSource(env);
+        KeyedStream<String, String> keyedStream = data.keyBy(new KeySelector<String, String>() {
+            @Override
+            public String getKey(String s) {
+                Map<String, Object> str2Map = GsonUtil.StringToMap(s);
+                for (Map.Entry<String, Object> entry : str2Map.entrySet()) {
+                    String key = entry.getKey();
+                    if (key.equals("name")) {
+                        return (String) entry.getValue();
+                    }
+                }
+                return "";
+            }
+        });
+
+        keyedStream
+                .timeWindow(Time.seconds(5))
+                .fold(new Tuple2<>("", 0L), new FoldFunction<String, Tuple2<String, Long>>() {
+                    @Override
+                    public Tuple2<String, Long> fold(Tuple2<String, Long> acc, String s) {
+                        Map<String, Object> str2Map = GsonUtil.StringToMap(s);
+                        for (Map.Entry<String, Object> entry : str2Map.entrySet()) {
+                            String key = entry.getKey();
+                            if (key.equals("name")) {
+                                acc.f0 = (String)entry.getValue();
+                            }else{
+                                acc.f1 += (Long)entry.getValue();
+                            }
+                        }
+                        return acc;
+                    }
+                });
         data.flatMap(new GroupCountSotFn())
                 .keyBy(0)
                 .sum(1);

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

@@ -2,17 +2,19 @@ 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.Tuple4;
 import org.apache.flink.api.java.tuple.Tuple5;
 
 public class BuildOutputTupleMapFunction extends RichMapFunction<
         Tuple5<Integer, Integer, Integer, Integer, Integer>,
-        Tuple3<Integer, Integer, Integer>> {
+        Tuple4<Integer,Integer, Integer, Integer>> {
     @Override
-    public Tuple3<Integer, Integer, Integer> map(Tuple5<Integer, Integer, Integer, Integer, Integer> inputTuple) throws Exception {
+    public Tuple4<Integer,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>(
+        return new Tuple4<Integer,Integer, Integer, Integer>(
                 inputTuple.f0,
                 inputTuple.f1,
+                inputTuple.f3,
                 inputTuple.f4);
     }
 }

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

+package com.byit.test;
+
+import com.byit.common.utils.GsonUtil;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.util.Collector;
+
+import java.util.Map;
+
+public class GroupCountSotFn implements FlatMapFunction<String, Tuple2<String, Double>> {
+
+    @Override
+    public void flatMap(String s, Collector<Tuple2<String, Double>> collector) throws Exception {
+        Map<String, Object> str2Map = GsonUtil.StringToMap(s);
+        for(Map.Entry<String, Object> entry :str2Map.entrySet()){
+            collector.collect(new Tuple2<String, Double>(entry.getKey(), (Double) entry.getValue()));
+        }
+    }
+}

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

+package com.byit.test;
+
+import com.byit.computer.common.MetricSchema;
+import com.byit.computer.constant.PropertiesConstants;
+import com.byit.computer.util.ExecutionEnvUtil;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.IterativeStream;
+import org.apache.flink.streaming.api.datastream.WindowedStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+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 java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+public class Test2 {
+    public static void main(String[] args) throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);
+        ExecutionEnvUtil.prepare(env,parameterTool);
+
+        DataStream<Tuple2<String, Double>> inputStream = buildSource(env);
+        IterativeStream<Tuple2<String, Double>> iterate = inputStream.iterate();
+
+
+        env.execute("Streaming Iteration Example");
+    }
+
+    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);
+    }
+
+    /**
+     * @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);
+    }
+
+    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;
+    }
+}