一、需求：

    1. 滑动窗口每20秒读取1分钟内数据，求平均值，最大值，最小值

    二、实现：

    1. pom文件 ```xml <?xml version=”1.0” encoding=”UTF-8”?> <project xmlns=”http://maven.apache.org/POM/4.0.0“

      1.   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      2.   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      4.0.0

      com.macro.flinkavg flinkavg

      0.0.1 
    1. <dependencies>
    3.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->
    4.     <dependency>
    5.         <groupId>org.apache.flink</groupId>
    6.         <artifactId>flink-connector-kafka_2.12</artifactId>
    7.         <version>1.9.0</version>
    8.     </dependency>
    11.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
    12.     <dependency>
    13.         <groupId>org.apache.flink</groupId>
    14.         <artifactId>flink-streaming-java_2.12</artifactId>
    15.         <version>1.9.0</version>
    16.     </dependency>
    19.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
    20.     <dependency>
    21.         <groupId>org.apache.flink</groupId>
    22.         <artifactId>flink-java</artifactId>
    23.         <version>1.9.0</version>
    24.     </dependency>
    26.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
    27.     <dependency>
    28.         <groupId>org.apache.flink</groupId>
    29.         <artifactId>flink-clients_2.12</artifactId>
    30.         <version>1.9.0</version>
    31.     </dependency>
    33.     <!--log4j2到slf4j桥梁-->
    34.     <dependency>
    35.         <groupId>org.apache.logging.log4j</groupId>
    36.         <artifactId>log4j-slf4j-impl</artifactId>
    37.         <version>2.9.1</version>
    38.     </dependency>
    39.     <!--kfka客户端-->
    40.     <dependency>
    41.         <groupId>org.apache.kafka</groupId>
    42.         <artifactId>kafka-clients</artifactId>
    43.         <version>2.2.1</version>
    44.     </dependency>
    45. </dependencies>
    46. <build>
    47.     <plugins>
    48.         <plugin>
    49.             <artifactId>maven-compiler-plugin</artifactId>
    50.             <configuration>
    51.                 <source>1.8</source>
    52.                 <target>1.8</target>
    53.                 <encoding>UTF-8</encoding>
    54.             </configuration>
    55.         </plugin>
    56.     </plugins>
    57. </build>

    2. 2. kafka配置文件

    bootstrap.servers=cdh1.macro.com:9092 acks=all retries=1 batch.size=16384 linger.ms=1 buffer.memory=33554432 key.serializer=org.apache.kafka.common.serialization.StringSerializer value.serializer=org.apache.kafka.common.serialization.StringSerializer topic=test

    2. 3. kafka生产者
    3. ```java
    4. package com.macro.flinkavg.util;
    6. import org.apache.kafka.clients.producer.KafkaProducer;
    7. import org.apache.kafka.clients.producer.ProducerRecord;
    9. import java.io.*;
    10. import java.util.Properties;
    12. public class SendToProducerUtill {
    14.     public static void main(String[] args) throws IOException {
    15.         Properties properties = new Properties();
    16.         FileInputStream fileInputStream = new FileInputStream(new File("E:\\JavaWorkSpace\\flinkavg\\src\\main\\resources\\ProducerConf.properties"));
    17.         properties.load(fileInputStream);
    18.         // 创建生产者对象
    19.         KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
    20.         // 发送数据
    21.         // 创建流对象
    22.         BufferedReader br = null;
    23.         try {
    24.             br = new BufferedReader(new FileReader("E:\\JavaWorkSpace\\flinkavg\\testData\\data"));//args[0]  //"datas/userInfo.txt"
    25.             // 定义字符串,保存读取的一行文字
    26.             String line = null;
    27.             // 循环读取,读取到最后返回null
    28.             while ((line = br.readLine()) != null) {
    29.                 System.out.println("成功发送消息："+line);
    30.                 producer.send(new ProducerRecord<String, String>(properties.getProperty("topic"), line));
    31.             }
    32.         } catch (Exception e) {
    33.             e.printStackTrace();
    34.         } finally {
    35.             // 释放资源
    36.             if (br != null) {
    37.                 try {
    38.                     br.close();
    39.                 } catch (IOException e) {
    40.                     e.printStackTrace();
    41.                 }
    42.             }
    43.         }
    45.         // 关闭资源
    46.         producer.close();
    47.     }
    48. }
    1. 实体类： ```java package com.macro.flinkavg.entity;

    public class PowerInfo { private Long time; private String fanName; private Double value; private String year; private String month; private String day; private Integer counts;

    1. public PowerInfo() { }
    3. public PowerInfo(Long time, String fanName, Double value, String year, String month, String day, Integer counts) {
    4.     this.time = time;
    5.     this.fanName = fanName;
    6.     this.value = value;
    7.     this.year = year;
    8.     this.month = month;
    9.     this.day = day;
    10.     this.counts = counts;
    11. }
    13. public Long getTime() {
    14.     return time;
    15. }
    17. public void setTime(Long time) {
    18.     this.time = time;
    19. }
    21. public String getFanName() {
    22.     return fanName;
    23. }
    25. public void setFanName(String fanName) {
    26.     this.fanName = fanName;
    27. }
    29. public Double getValue() {
    30.     return value;
    31. }
    33. public void setValue(Double value) {
    34.     this.value = value;
    35. }
    37. public String getYear() {
    38.     return year;
    39. }
    41. public void setYear(String year) {
    42.     this.year = year;
    43. }
    45. public String getMonth() {
    46.     return month;
    47. }
    49. public void setMonth(String month) {
    50.     this.month = month;
    51. }
    53. public String getDay() {
    54.     return day;
    55. }
    57. public void setDay(String day) {
    58.     this.day = day;
    59. }
    61. public Integer getCounts() {
    62.     return counts;
    63. }
    65. public void setCounts(Integer counts) {
    66.     this.counts = counts;
    67. }
    69. @Override
    70. public String toString() {
    71.     return "PowerInfo{" +
    72.             "time=" + time +
    73.             ", fanName='" + fanName + '\'' +
    74.             ", value=" + value +
    75.             ", year='" + year + '\'' +
    76.             ", month='" + month + '\'' +
    77.             ", day='" + day + '\'' +
    78.             ", counts=" + counts +
    79.             '}';
    80. }

    }

    2. 5. 逻辑代码
    3. ```java
    4. package com.macro.flinkavg.flink;
    6. import com.macro.flinkavg.entity.PowerInfo;
    7. import org.apache.flink.api.common.serialization.SimpleStringSchema;
    8. import org.apache.flink.api.java.tuple.Tuple;
    9. import org.apache.flink.api.java.tuple.Tuple4;
    10. import org.apache.flink.streaming.api.TimeCharacteristic;
    11. import org.apache.flink.streaming.api.datastream.DataStream;
    12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    13. import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
    14. import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    15. import org.apache.flink.streaming.api.windowing.time.Time;
    16. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    17. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
    18. import org.apache.flink.util.Collector;
    20. import java.text.ParsePosition;
    21. import java.text.SimpleDateFormat;
    22. import java.util.ArrayList;
    23. import java.util.Iterator;
    24. import java.util.Properties;
    26. public class VestoreDataAVGByFlink {
    28.     public static void main(String[] args) throws Exception {
    29.         // 1. 创建环境
    30.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    31.         env.setParallelism(1);
    32.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    33.         // 2. 读取kafka数据，创建 DataStream
    34.         Properties properties = new Properties();
    35.         properties.setProperty("bootstrap.servers", "cdh1.macro.com:9092");
    36.         properties.setProperty("group.id", "test_group");
    37.         properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    38.         properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    39.         properties.setProperty("auto.offset.reset", "latest");
    40.         DataStream<String> inputStream = env.addSource(new FlinkKafkaConsumer<String>("test", new SimpleStringSchema(), properties));
    42.         // 3. 转换为 PowerInfo，并分配时间戳和 watermark
    43.         DataStream<PowerInfo> dataStream = inputStream
    44.                 .map(line -> {
    45.                     String[] fields = line.split(",");
    46.                     Long timeSimple = (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")).parse(fields[0], new ParsePosition(0)).getTime();
    47.                     return new PowerInfo(timeSimple, fields[1], new Double(fields[2]), fields[3], fields[4], fields[5], 1);
    48.                 })
    49.                 .assignTimestampsAndWatermarks(
    50.                         new AscendingTimestampExtractor<PowerInfo>() {
    51.                             @Override
    52.                             public long extractAscendingTimestamp(PowerInfo timeStamp) {
    54.                                 return timeStamp.getTime();
    55.                             }
    56.                         });
    58.         // 4. 分组开窗聚合，得到每个窗口内value的最大值最小值和平均值
    59.         dataStream.keyBy("counts")
    60.                 .timeWindow(Time.minutes(1), Time.seconds(20))
    61.                 .apply( new VerstoreAggregate()).print();
    64.         // 执行
    65.         env.execute();
    67.     }
    69.     // 全窗口聚合
    70.     public static class VerstoreAggregate implements WindowFunction<PowerInfo, Tuple4<Long, Double, Double, Double>, Tuple, TimeWindow> {
    72.         ArrayList<Double> arrayList = new ArrayList<>();
    74.         @Override
    75.         public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<PowerInfo> iterable, Collector<Tuple4<Long, Double, Double, Double>> collector) throws Exception {
    77.             Iterator<PowerInfo> iterator = iterable.iterator();
    78.             while (iterator.hasNext()) {
    79.                 Double value = iterator.next().getValue();
    80.                 arrayList.add(value);
    81.             }
    83.             Double sum = 0.0;
    84.             Double max = 0.0;
    85.             Double min = 0.0;
    87.             // 判断列表非空，以第一个元素作为最大值和最小值
    88.             if ( ! arrayList.isEmpty()) {
    89.                 max = arrayList.get(0);
    90.                 min = arrayList.get(0);
    91.             }
    93.             // 循环遍历列表，取最大值最小值和和
    94.             for (Double aDouble : arrayList) {
    95.                 //System.out.println("value:"+aDouble);
    96.                 sum += aDouble;
    97.                 if (max < aDouble) {
    98.                     max = aDouble;
    99.                 }
    100.                 if (min > aDouble) {
    101.                     min = aDouble;
    102.                 }
    103.             }
    105.             //System.out.println("count:"+arrayList.size());
    106.             //System.out.println("{max:" + max + " min:" + min + " sum:" + sum + " endTime:" + timeWindow.getEnd()+"}");
    107.             // 返回窗口内窗口结束时间，最大值，最小值，和平均值。
    108.             collector.collect(new Tuple4<>(timeWindow.getEnd(),max,min,sum/arrayList.size()));
    109.             arrayList.clear();
    110.         }
    111.     }
    113. }

    三、结果展示

    1. kafka生产数据

    image.png

    1. flink消费数据，展示结果

    33ad04bbf304b290514422e41250b83.png