【大数据】Flink学习笔记

认识Flink

Docker安装Flink

version: "2.1"
services:
  jobmanager:
    image: flink
    expose:
      - "6123"
    ports:
      - "20010:8081"
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
  taskmanager:
    image: flink
    expose:
      - "6121"
      - "6122"
    depends_on:
      - jobmanager
    command: taskmanager
    links:
      - "jobmanager:jobmanager"
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager

docker-compose up -d

基本概念

1. 核心目标: 数据流上有状态的计算 Stateful computation over data stream
2. Flink: 一个框架和分布式处理引擎, 对有界和无界的数据流进行有状态计算 Flink是事件驱动型的应用
3. 有界数据流: 有定义流的开始, 但没有定义流的结束; 而有界数据流即有流的开始, 也有结束
4. 状态: 把流处理需要的而外数据保存成一个状态, 然后针对这条数据处理, 并更新状态

Flink的特点

1. 高吞吐低延迟: 每秒处理百万个事件, 毫秒级延迟
2. 结果的准确性: Flink提供了事件时间和处理时间的语义, 对乱序事件流, 事件事件语义热能能够提供一致且准确的结果
3. 精确一次: 精确一次的状态一致性保证 不丢数, 不重复
4. 可以连接到常用的存储系系统, 如Kafka, JDBC, redis, Hive, HDFS等
5. 高可用: 本身具有高可用设置, 且可以与K8s/YARN/Mesos紧密集成

Flink 和 Spark Streaming 对比

基本使用

WordCount实现

依赖

        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>1.17.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>1.17.1</version>
        </dependency>

批模式代码

public class WordCount {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 读取文件
        DataSource<String> lines = env.readTextFile("bigdata/src/test/resources/word list.txt");

        // 文本拆分, 转化为词频元组
        lines.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (line, out) -> {
                    // 按照空格切分单词
                    Arrays.stream(line.split(" "))
                            .map(word -> Tuple2.of(word, 1)) // 将单词转化为词频元组
                            .forEach(out::collect); // 向下游发送数据
                }).returns(new TypeHint<>() { // 需要有这个, 否则会报错
                }).groupBy(0) // 根据元组的索引0元素进行分组
                .sum(1) // 根据元组的索引1元素求和
                .print();
    }
}

输出

# 这里去除了一些日志
(passnight,2)
(hadoop,2)
(like,2)
(love,2)
(I,4)

流模式代码

public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        // 创建流式处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从文件读取数据
        env.readTextFile("bigdata/src/test/resources/word list.txt")
                // 将数据分词, 并转化为词频二元组
                .flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (line, out) -> Arrays.stream(line.split(" "))
                        .map(word -> Tuple2.of(word, 1)) // 转化为词频二元组
                        .forEach(out::collect)) // 将数据传给下游
                .returns(new TypeHint<>() {
                    // 根据key进行分组
                }).keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
                .sum(1) // 分组后根据索引为0的元素进行聚合
                .print();
        env.execute(); // 启动任务
    }
}

输出

17> (like,1)
11> (passnight,1)
6> (I,1)
11> (passnight,2)
13> (love,1)
6> (I,2)
6> (I,3)
18> (hadoop,1)
18> (hadoop,2)
13> (love,2)
6> (I,4)
17> (like,2)

可见不同线程以流的方式一条一条处理文件中的数据

网络流模式代码

package com.passnight.bigdata.flink;

import lombok.Cleanup;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.functions.KeySelector;
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;

import java.util.Arrays;

public class SocketStreamWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("server.passnight.local", 30000)
                .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> Arrays.stream(line.split(" "))
                        .map((String word) -> Tuple2.of(word, 1))
                        .forEach(out::collect)).returns(new TypeHint<>() {
                }).keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
                .sum(1)
                .print();
        env.execute();
    }
}

启动netcat生成文本流

passnight@passnight-s600:~$ netcat -lk 30000

输出

# netcat中输入 "hello world"
6> (hello,1)
11> (world,1)
# 输入 "hello flink"
16> (flink,1)
6> (hello,2)

在web UI上提交代码

创建项目¹

mvn archetype:generate                               \
      -DarchetypeGroupId=org.apache.flink              \
      -DarchetypeArtifactId=flink-quickstart-java      \
      -DarchetypeVersion=1.4.2

将Flink版本修改到``与上面匹配

		<flink.version>1.7.1</flink.version>

编写代码

package com.passnight.bigdata.flink;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

public class SocketStreamWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("server.passnight.local", 30000)
                .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> Arrays.stream(line.split(" "))
                        .map((String word) -> Tuple2.of(word, 1))
                        .forEach(out::collect)).returns(new TypeHint<>() {
                }).keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
                .sum(1)
                .print();
        env.execute();
    }
}

配置打包

<mainClass>com.passnight.monitor.SocketStreamWordCount</mainClass>

将启动类设置成我们的启动类

在Web UI上提交

Flink 架构

系统架构

执行流程:

1. 客户端解析提供的参数, 将其封装为一个任务, 发送给JobManager
2. JobManager会将任务提交到JobMaster
3. JobMaster之后会向资源管理器申请资源
4. 资源管理器之后会向TaskManager发送请求, 在可以接受请求, TaskManager会返回信息, 这样资源管理器就可以分发任务

核心概念

并行度

并行度: 同时独立被执行的任务数 一个流程序的并行度, 一般被认为所有算子最大的并行度

1. 大数据环境下, 可以将一个算子复制到多个节点, 这样一个算子就被拆成了多个子任务, 由不同节点并行执行

2. 这些子任务在不同线程/不同物理机/不同容器中完全独立运行

3. 下图是并行度为1和并行度为2的区别:

4. 所有的算子的并行度都可以通过setParalism命令配置

5. 实验:

   1. 添加以下依赖, 就可以在单机模式中看到web UI:

              <dependency>
                  <groupId>org.apache.flink</groupId>
                  <artifactId>flink-runtime-web</artifactId>
                  <version>1.17.1</version>
              </dependency>
      

   2. 并且调用setParalism将flatMap算子的并行度设置为2

      class ParallelSocketStreamWordCount {
          public static void main(String[] args) throws Exception {
              @Cleanup
              StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
              env.socketTextStream("server.passnight.local", 30000)
                      .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> Arrays.stream(line.split(" "))
                              .map((String word) -> Tuple2.of(word, 1))
                              .forEach(out::collect)).returns(new TypeHint<>() {
                      }).setParallelism(2) // 设置并行度为2
                      .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
                      .sum(1)
                      .print();
              env.execute();
          }
      }
      

   3. 可以在Web UI中看到结果:

   4. 这里也可以看到默认的并行度是核心数

   5. 同样的可以全局设置, 通过调用env.envParallelism()设置, 也可以在提交的时候设置参数-p来配置webUI页面也有对应的参数配置框; 算子的优先级遵循越局部越有限的原则

算子链(Opeartor Chain)

1. 数据可能由多个算子操作, 这些算子会形成算子链

2. 数据流在算子之间的传输形式可能会死一对一的直通模式, 也可能是打乱的重分区模式

   1. 一对一: 这种模式, 算子读取数据之后, 直接发送给map算子做处理, 不需要经过重分区, 也不需要**调整数据的顺序 ** 类似Spark的窄依赖
   2. 重分区: 这会根据数据流的特征发生变化, 如keyBy操作, 会根据key的特征对数据重分区; Flink会根据数据传输策略, 把数据发送给不同的下游任务 除了forward都是重分区

3. 合并算子链: 并行度相同的一对一操作, 可以被合并成一个大任务 这是Flink自带的优化, 可以减少IO负载, 及内存交换负载

   1. 有两种方式禁用这样的优化, 分别是通过disableOperatorChanning和disableChanningapi; 也可以使用startNewChain开启新链条, 以达到类似禁用的效果

   2. 默认情况下, map和flatenMap被分为一组, keyBy和print被分为一组:

   3. 在调用env.disableOperatorChaining();后, 可以全局禁止任务合并:

   4. 或是调用disableChanning禁止某一个算子的合并:

   5. 还可以通过创建新链达到类似的效果startNewChain:

   6. 代码:

      public class OperatorChain {
          public static void main(String[] args) throws Exception {
              @Cleanup
              StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
      
      //        env.disableOperatorChaining(); // 全局禁用算子链
              env.socketTextStream("server.passnight.local", 30000)
                      .flatMap((String line, Collector<String> out) ->
                              Arrays.stream(line.split(" "))
                                      .forEach(out::collect))
                      .returns(new TypeHint<>() {
                      }) // .disableChaining() 局部禁用算子链
                      .map((String word) -> Tuple2.of(word, 1))
                      .startNewChain() // 开启一个新的算子链
                      .returns(new TypeHint<>() {
                      }).keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
                      .sum(1)
                      .print();
              env.execute();
          }
      }
      
      

任务槽

1. 一个TaskManager的计算资源是有限的, 并行的任务越多, 每个线程的资源越少. 为了控制并发量, 任务槽对每个任务所占用的资源做出了划分. 任务槽表示TaskManager拥有资源的固定大小; 其资源分配主要针对内存
2. 一般来说, 任务槽的数量是CPU的核心数; 具体可以通过taskmanager.numberOfTaskSlots=1进行配置 slot划分内存, 但是不划分CPU
3. 同一个作业中, 同一个slot上, 不同算子的子任务是同时运行的 默认情况下, 同一个slot共享组的slot是共享的(默认都是Default组), 但也可以通过slotSharingGroup配置
4. slot和并行度: slot是静态的概念, 而并行度是动态概念; 并行度指实际上同时进行的任务数, 一般来说slot决定并行度的上限; 上图souce+map算子链和keyby+window+apply算子链的并行度为6, sink的并行度为1 flink不支持并行度大于任务槽的情况, 提交会抛出NoResourceAvaiableException异常
5. Yarn模式下会动态申请TM, 其数量为$申请TM数量=\lceil \frac{job并行度}{每个TMslot数}\rceil$

Flink作业提交流程

Standalone会话模式提交流程

1. 客户端会根据参数生成逻辑流图, 再生成作业流图 这个过程会进行算子优化

2. 然后向JobManager提交, 生成执行图

3. 之后资源管理器会向TaskManager请求并分配任务, 此时TaskManager会生成物理流图

4. 各种图的创建顺序为:

YARN应用模式作业提交流程

1. 向YARN的Resource Manager提交节点, 然后在NodeManager中启动ApplicationMaster的容器
2. 之后ApplicationMaster会启动资源管理器和分发器; JobMaster生成逻辑流图, 再生成作业流图/执行流图
3. 生成了执行流图后会向资源管理器申请/请求slot资源;
4. 该资源管理器会向YARN的ResourceManager申请资源, ResourceManager再实际分配资源, 并返回给Flink的资源管理器

DataStream API

源算子

从集合中输入

class ReadFromCollection {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromCollection(IntStream.range(0, 10)
                        .boxed()
                        .collect(Collectors.toUnmodifiableList()))
                .print();
        env.execute();
    }
}

输出结果为:

4> 0
6> 2
7> 3
5> 1
10> 6
9> 5
8> 4
11> 7
12> 8
13> 9

从文件读取

从文件读取需要添加相应的connector

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-files</artifactId>
            <version>1.17.1</version>
        </dependency>

代码为

class ReadFromFile {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(),
                new Path("bigdata/src/main/resources/word list.txt")).build();
        env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "fileSource")
                .print();
        env.execute();
    }
}

输出为:

12> I love passnight
12> I like passnight
12> I love hadoop
12> I like hadoop

从Kafka中读取

从kafka中读取需要导入依赖 (注意:flink-connector-kafka中依赖的kafka-client版本太低, 需要手动导入更高版本的kafka-client)

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>1.17.1</version>
    </dependency>	
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.6.0</version>
        </dependency>

代码为

class ReadFromKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("server.passnight.local:20015") // 指定连接url
                .setGroupId("testGroup") // 指定消费者组
                .setTopics("test") // 指定topic
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定反序列化其
                .setStartingOffsets(OffsetsInitializer.latest()) // 指定消费偏移量
                .build();

        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource")
                .print();
        env.execute();
    }
}

从数据生成器读取

从数据生成器中读取需要导入依赖

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-datagen</artifactId>
            <version>1.17.1</version>
        </dependency>

代码为:

class ReadFromDataGenerator {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource<>(
                (GeneratorFunction<Long, String>) aLong -> String.format("Id: %d", aLong), // 生成器方法, 主要重写里面的map
                10, // 自动生成的序列号的最大值, 达到这个值之后就停止生成
                RateLimiterStrategy.perSecond(1), // 限速, 限制每秒钟生成的数据量
                Types.STRING);

        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "dataGenerator")
                .print();

        env.execute();
    }
}

输出为:

6> Id: 5
14> Id: 7
9> Id: 6
4> Id: 2
20> Id: 0
2> Id: 1
18> Id: 4
3> Id: 3
8> Id: 9
15> Id: 8

可以看到一共生成了10条数据就停止了; 且限速也成功生效

映射算子

用于测试的pojo

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class WaterSensor {
    private String id;
    private Long ts;
    private Integer vc;
}

Map算子

class Map {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromCollection(IntStream.range(0, 10)
                        .boxed()
                        .map(integer -> new WaterSensor(String.format("s%d", integer), Long.valueOf(integer), integer))
                        .collect(Collectors.toUnmodifiableList()))
                .map(WaterSensor::getId)
                .print();

        env.execute();
    }
}

输出为:

19> s4
20> s5
18> s3
1> s6
17> s2
4> s9
15> s0
16> s1
3> s8
2> s7

可以看到已经将WaterSensor的id转换并输出了

Filter算子

class Filter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromCollection(IntStream.range(0, 10)
                        .boxed()
                        .map(integer -> new WaterSensor(String.format("s%d", integer), Long.valueOf(integer), integer))
                        .collect(Collectors.toUnmodifiableList()))
                .filter(waterSensor -> waterSensor.getTs() > 5)
                .print();

        env.execute();
    }
}

输出为:

10> WaterSensor(id=s9, ts=9, vc=9)
7> WaterSensor(id=s6, ts=6, vc=6)
8> WaterSensor(id=s7, ts=7, vc=7)
9> WaterSensor(id=s8, ts=8, vc=8)

FlatMap算子

class FlatMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromCollection(Arrays.asList("1 2 3 4", "5 6 7", "8 9"))
                .flatMap((FlatMapFunction<String, String>) (value, out) -> Arrays.stream(value.split(" ")).forEach(out::collect))
                .returns(Types.STRING)
                .print();

        env.execute();
    }
}

输出为:

10> 1
11> 5
10> 2
10> 3
10> 4
11> 6
12> 8
12> 9
11> 7

聚合算子

1. 在Flink当中, 若要做聚合, 先要对数据进行分区, 该操作就是通过keyBy算子完成的
2. 在相同key的数据被发送到一个分区后, 对该分区的数据计算, 便完成了聚合操作

KeyBy算子

class KeyBy {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                        new WaterSensor("s1", 1L, 1),
                        new WaterSensor("s1", 11L, 11),
                        new WaterSensor("s2", 2L, 2),
                        new WaterSensor("s3", 3L, 3)
                )
                .keyBy(WaterSensor::getId)
                /*
                 * 调用keyBy后返回`KeyedStream`, 之后就有一系列聚合函数
                 * 它在执行流图中体现为**箭头**而非流程框
                 */
                .print()
                .setParallelism(2);

        env.execute();
    }
}

输出为:

2> WaterSensor(id=s1, ts=1, vc=1)
1> WaterSensor(id=s2, ts=2, vc=2)
2> WaterSensor(id=s1, ts=11, vc=11)
2> WaterSensor(id=s3, ts=3, vc=3)

可以看到相同Key的数据在同一分区

简单聚合算子

class SimpleAggregation {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                        new WaterSensor("s1", 1L, 1),
                        new WaterSensor("s1", 11L, 11),
                        new WaterSensor("s2", 2L, 2),
                        new WaterSensor("s3", 3L, 3)
                )
                .keyBy(WaterSensor::getId)
//                .sum("vc")
//                .min("vc")
//                .max("vc")
                .maxBy("vc")
                .print();

        env.execute();
    }
}

输出结果为

####################### sum算子
18> WaterSensor(id=s3, ts=3, vc=3)
20> WaterSensor(id=s1, ts=1, vc=1)
5> WaterSensor(id=s2, ts=2, vc=2)
20> WaterSensor(id=s1, ts=1, vc=12) # 第二次出现"s1", 值=1+11=12
##################### min算子
20> WaterSensor(id=s1, ts=1, vc=1)
5> WaterSensor(id=s2, ts=2, vc=2)
18> WaterSensor(id=s3, ts=3, vc=3)
20> WaterSensor(id=s1, ts=1, vc=1) # 第二次出现"s1", 第一次出现的更小, 因此还是1
#################### max算子
18> WaterSensor(id=s3, ts=3, vc=3)
20> WaterSensor(id=s1, ts=1, vc=1)
5> WaterSensor(id=s2, ts=2, vc=2)
20> WaterSensor(id=s1, ts=1, vc=11) # 第二次出现"s1", 第2次出现的更大, 因此还是11
################## maxBy算子, max只会取比较字段的最大值, 非比较字段保留; 而maxBy非比较字段也会取比较字段的最大值
5> WaterSensor(id=s2, ts=2, vc=2)
20> WaterSensor(id=s1, ts=1, vc=1)
18> WaterSensor(id=s3, ts=3, vc=3)
20> WaterSensor(id=s1, ts=11, vc=11) # 可以看到ts也变成了11

reduceBy算子

class Reduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                        new WaterSensor("s1", 1L, 1),
                        new WaterSensor("s1", 11L, 11),
                        new WaterSensor("s2", 2L, 2),
                        new WaterSensor("s3", 3L, 3)
                )
                .keyBy(WaterSensor::getId)
                .reduce((x, y) -> new WaterSensor(x.getId(), y.getTs(), x.getVc() + y.getVc()))
                .print();

        env.execute();
    }
}

输出结果为:

20> WaterSensor(id=s1, ts=1, vc=1)
5> WaterSensor(id=s2, ts=2, vc=2)
18> WaterSensor(id=s3, ts=3, vc=3)
20> WaterSensor(id=s1, ts=11, vc=12)

可以看到vc执行了求和操作; 而ts每次都取后者 注意每个组的第一条数据不会进入聚合方法

UDF

用户可以实现Flink提供的接口实现自定义函数

class UDF {
    public static class MyFilterFunction implements FilterFunction<WaterSensor> {

        private static final String id = "s1";

        @Override
        public boolean filter(WaterSensor value) {
            return Objects.equals(id, value.getId());
        }
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                        new WaterSensor("s1", 1L, 1),
                        new WaterSensor("s1", 11L, 11),
                        new WaterSensor("s2", 2L, 2),
                        new WaterSensor("s3", 3L, 3)
                )
                .filter(new MyFilterFunction())
                .print();

        env.execute();
    }
}

输出结果为:

4> WaterSensor(id=s1, ts=1, vc=1)
5> WaterSensor(id=s1, ts=11, vc=11)

富函数

在执行算子的时候, 可以使用富函数以添加生命周期函数及获取运行时信息

class RichFunction {
    public static void main(String[] args) throws Exception {
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.fromElements(
                        new WaterSensor("s1", 1L, 1),
                        new WaterSensor("s1", 11L, 11),
                        new WaterSensor("s2", 2L, 2),
                        new WaterSensor("s3", 3L, 3)
                )
                .map(new RichMapFunction<WaterSensor, Integer>() {
                    @Override
                    public Integer map(WaterSensor value) throws Exception {
                        return value.getVc();
                    }

                    // 富函数带的钩子
                    @Override
                    public void open(Configuration parameters) {
                        RuntimeContext context = getRuntimeContext();
                        System.out.printf("[open()] 子任务编号: %s, 子任务名称: %s%n",
                                context.getIndexOfThisSubtask(),
                                context.getTaskNameWithSubtasks());
                    }

                    @Override
                    public void close() {
                        RuntimeContext context = getRuntimeContext();
                        System.out.printf("[close()] 子任务编号: %s, 子任务名称: %s%n",
                                context.getIndexOfThisSubtask(),
                                context.getTaskNameWithSubtasks());
                    }
                })
                .print();

        env.execute();
        TimeUnit.DAYS.sleep(1);
    }
}

输出为:

[open()] 子任务编号: 10, 子任务名称: Map -> Sink: Print to Std. Out (11/20)#0
[open()] 子任务编号: 4, 子任务名称: Map -> Sink: Print to Std. Out (5/20)#0
[open()] 子任务编号: 12, 子任务名称: Map -> Sink: Print to Std. Out (13/20)#0
[open()] 子任务编号: 11, 子任务名称: Map -> Sink: Print to Std. Out (12/20)#0
[open()] 子任务编号: 9, 子任务名称: Map -> Sink: Print to Std. Out (10/20)#0
[open()] 子任务编号: 7, 子任务名称: Map -> Sink: Print to Std. Out (8/20)#0
[open()] 子任务编号: 16, 子任务名称: Map -> Sink: Print to Std. Out (17/20)#0
[open()] 子任务编号: 15, 子任务名称: Map -> Sink: Print to Std. Out (16/20)#0
[open()] 子任务编号: 3, 子任务名称: Map -> Sink: Print to Std. Out (4/20)#0
[open()] 子任务编号: 8, 子任务名称: Map -> Sink: Print to Std. Out (9/20)#0
[open()] 子任务编号: 5, 子任务名称: Map -> Sink: Print to Std. Out (6/20)#0
[open()] 子任务编号: 18, 子任务名称: Map -> Sink: Print to Std. Out (19/20)#0
[open()] 子任务编号: 19, 子任务名称: Map -> Sink: Print to Std. Out (20/20)#0
[open()] 子任务编号: 6, 子任务名称: Map -> Sink: Print to Std. Out (7/20)#0
[open()] 子任务编号: 13, 子任务名称: Map -> Sink: Print to Std. Out (14/20)#0
[open()] 子任务编号: 14, 子任务名称: Map -> Sink: Print to Std. Out (15/20)#0
[open()] 子任务编号: 2, 子任务名称: Map -> Sink: Print to Std. Out (3/20)#0
[open()] 子任务编号: 17, 子任务名称: Map -> Sink: Print to Std. Out (18/20)#0
[open()] 子任务编号: 1, 子任务名称: Map -> Sink: Print to Std. Out (2/20)#0
[open()] 子任务编号: 0, 子任务名称: Map -> Sink: Print to Std. Out (1/20)#0
16> 2
14> 1
15> 11
17> 3
[close()] 子任务编号: 9, 子任务名称: Map -> Sink: Print to Std. Out (10/20)#0
[close()] 子任务编号: 7, 子任务名称: Map -> Sink: Print to Std. Out (8/20)#0
[close()] 子任务编号: 0, 子任务名称: Map -> Sink: Print to Std. Out (1/20)#0
[close()] 子任务编号: 1, 子任务名称: Map -> Sink: Print to Std. Out (2/20)#0
[close()] 子任务编号: 5, 子任务名称: Map -> Sink: Print to Std. Out (6/20)#0
[close()] 子任务编号: 10, 子任务名称: Map -> Sink: Print to Std. Out (11/20)#0
[close()] 子任务编号: 6, 子任务名称: Map -> Sink: Print to Std. Out (7/20)#0
[close()] 子任务编号: 8, 子任务名称: Map -> Sink: Print to Std. Out (9/20)#0
[close()] 子任务编号: 2, 子任务名称: Map -> Sink: Print to Std. Out (3/20)#0
[close()] 子任务编号: 3, 子任务名称: Map -> Sink: Print to Std. Out (4/20)#0
[close()] 子任务编号: 4, 子任务名称: Map -> Sink: Print to Std. Out (5/20)#0
[close()] 子任务编号: 12, 子任务名称: Map -> Sink: Print to Std. Out (13/20)#0
[close()] 子任务编号: 13, 子任务名称: Map -> Sink: Print to Std. Out (14/20)#0
[close()] 子任务编号: 16, 子任务名称: Map -> Sink: Print to Std. Out (17/20)#0
[close()] 子任务编号: 11, 子任务名称: Map -> Sink: Print to Std. Out (12/20)#0
[close()] 子任务编号: 15, 子任务名称: Map -> Sink: Print to Std. Out (16/20)#0
[close()] 子任务编号: 14, 子任务名称: Map -> Sink: Print to Std. Out (15/20)#0
[close()] 子任务编号: 17, 子任务名称: Map -> Sink: Print to Std. Out (18/20)#0
[close()] 子任务编号: 19, 子任务名称: Map -> Sink: Print to Std. Out (20/20)#0
[close()] 子任务编号: 18, 子任务名称: Map -> Sink: Print to Std. Out (19/20)#0

可以注意到:

1. 一共调用了20次open和close, 因为默认的并行度为20
2. open和close相当于生命周期函数, 每个子任务启动和关闭时会调用一次
3. 在富函数中可以获取运行时上下文中的一些信息

分区算子

测试代码:

public class Partition {
    public static Partitioner<String> partitioner = (key, numPartitions) -> Integer.parseInt(key) % numPartitions;
    public static KeySelector<String, String> keySelector = value -> value;

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2)
                .socketTextStream("server.passnight.local", 30000)
//                .shuffle() // 随机分区
//                .rebalance() // 轮询算法
//                .rescale()
//                .broadcast() // 广播分区
//                .global() // 全局分区
                .partitionCustom(Partition.partitioner, Partition.keySelector)// 自定义分区
                .print();
        env.execute();
    }
}

shuffle分区

输出结果为:

2> 1
1> 1
2> 1
1> 1
2> 1
2> 1
2> 1
2> 1
2> 1

可以看到分区没有任何规律, 因为选择分区是随机的"如下代码", 因为分区算法为: return random.nextInt(numberOfChannels);

rebalance分区

输出结果为

1> 1
2> 1
1> 1
2> 1
1> 1
2> 1
# 

可以看到分区交替出现; 因为选自分区的时如下代码; 这样可以解决固定分区读取kafka数据倾斜的问题; 因为分区算法为: nextChannelToSendTo = (nextChannelToSendTo + 1) % numberOfChannels;

rescale分区

输出结果为:

2> 1
1> 1
2> 1
1> 1
2> 1

类似rebalance它会局部组队, 然后在组内轮询. 因为它只会和部分下游算子组成通道

boardcast分区

输出结果为:

# 发送了1, 2, 3
2> 1
1> 1
1> 2
2> 2
1> 3
2> 3

可以看到, 所有的分区都接收到了数据

global分区

输出结果为:

1> 1
1> 1
1> 1
1> 1
1> 1
1> 1

将使用全局分区算子之后, 所有的数据都只会发送到第一个分区; 所有的分区号都为1

自定义分区器

自定义分区器:

    public static Partitioner<String> partitioner = (key, numPartitions) -> Integer.parseInt(key) % numPartitions;
    public static KeySelector<String, String> keySelector = value -> value;

输出为:

2> 1
2> 1
1> 2
1> 2
2> 3
2> 3
1> 4
1> 4

可以看到, 数据选择的流和其值符合Integer.parseInt(key) % numPartitions方法计算的结果

分流

分流算子可以将一条数据流转化为完全独立的两条

下面是对数字分流的一个例子: 将数字流划分为奇数流和偶数流

Filter实现分流

可以对数据流分别使用不同的条件进行过滤, 数据被过滤器给过滤成了两个流; 这样实现性能较差, 因为每个过滤器都要处理所有的数据

class SplitByFilter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        DataStreamSource<String> dataStream = env.socketTextStream("server.passnight.local", 30000);

        // 使用filter分流
        dataStream.filter(x -> Integer.parseInt(x) % 2 == 0).print("EvenStream");
        dataStream.filter(x -> Integer.parseInt(x) % 2 == 1).print("OddStream");
        env.execute();
    }
}

输出为:

# 输入1-5的数字
OddStream:5> 1
EvenStream:17> 2
OddStream:7> 3
EvenStream:19> 4
OddStream:9> 5

从webUI中可以看到已经分成两个流了

侧输出流实现分流

假设有程序的上下文, 便可以调用上下文的output; 这样就可以输出任何类型的数据到任意流中了

class SplitBySideOutput {
    public static void main(String[] args) throws Exception {
        final OutputTag<WaterSensor> stream1Tag = new OutputTag<>("stream1", Types.POJO(WaterSensor.class));
        final OutputTag<WaterSensor> stream2Tag = new OutputTag<>("stream2", Types.POJO(WaterSensor.class));
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        SingleOutputStreamOperator<WaterSensor> process = env.socketTextStream("server.passnight.local", 30000)
                .map(value -> new WaterSensor("s" + Integer.parseInt(value) % 3, Long.parseLong(value), Integer.parseInt(value)))
                .process(new ProcessFunction<>() {
                    @Override
                    public void processElement(WaterSensor waterSensor, ProcessFunction<WaterSensor, WaterSensor>.Context ctx, Collector<WaterSensor> out) throws Exception {
                        if (waterSensor.getVc() < 5 && waterSensor.getVc() > 0) {
                            // 当 waterSensor.vc in (0,5)时走这条支流
                            ctx.output(stream1Tag, waterSensor);
                        } else if (waterSensor.getVc() > 5 && waterSensor.getVc() < 10) {
                            // 当 waterSensor.vc in (5,10)时走这条支流
                            ctx.output(stream2Tag, waterSensor);
                        } else {
                            // 默认走主流
                            out.collect(waterSensor);
                        }
                    }
                });
        process.print(); // 打印主流的数据

        // 从主流中获取测流数据
        process.getSideOutput(stream1Tag).print("侧流1");
        process.getSideOutput(stream2Tag).print("侧流2");


        env.execute();
    }
}

输出结果为:

输入为: 8,9,10,11,12
侧流1:8> WaterSensor(id=s1, ts=1, vc=1)
9> WaterSensor(id=s1, ts=10, vc=10)
侧流2:10> WaterSensor(id=s2, ts=8, vc=8)
11> WaterSensor(id=s2, ts=5, vc=5)
侧流1:12> WaterSensor(id=s1, ts=1, vc=1)

合流

使用union合流

class CombineByUnion {
    public static void main(String[] args) throws Exception {
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<Integer> dataStream1 = env.fromElements(1, 2, 3, 4, 5, 6);
        DataStreamSource<Integer> dataStream2 = env.fromElements(11, 22, 33, 44, 55, 66);
        DataStreamSource<String> dataStream3 = env.fromElements("111", "222", "333", "444", "555", "666");

        dataStream1.union(dataStream2)
                .union(dataStream3.map(Integer::parseInt))
                .print();

        env.execute();
    }
}

输出结果为:

4> 11
5> 22
6> 33
7> 44
7> 111
8> 55
8> 222
9> 66
9> 333
10> 444
11> 555
12> 666
13> 1
14> 2
15> 3
16> 4
17> 5
18> 6

Connect

Union只支持相同数据类型的合流, 而Connect支持不同数据类型的合流, 但是它得到的不是DataStream而是连接流, 它们形式上再一个流中, 但数据时分开的.

class CombineByConnect {
    public static class Map {
        public static void main(String[] args) throws Exception {
            @Cleanup
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

            DataStreamSource<Integer> dataStream1 = env.fromElements(1, 2, 3);
            DataStreamSource<String> dataStream3 = env.fromElements("a", "b", "c");

            // connect支持类型不一样的数据类型的流
            // connect只能执行一次
            dataStream1.connect(dataStream3)
                    .map(new CoMapFunction<Integer, String, String>() {

                        @Override
                        public String map1(Integer value) {
                            return String.valueOf(value * 10);
                        }

                        @Override
                        public String map2(String value) {
                            return value;
                        }
                    }).print();


            env.execute();
        }
    }
}

输出结果为:

20> a
10> 10
2> c
11> 20
1> b
12> 30

输出算子(sink)

输出到文件系统

class FileSinkOutput {
    public static void main(String[] args) throws Exception {
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        DataGeneratorSource<Long> dataSource = new DataGeneratorSource<>(x -> x, Long.MAX_VALUE, RateLimiterStrategy.perSecond(1000), Types.LONG);

        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE); // 必须开启CheckPoint, 否则文件一直都处于in progress状态
        env.fromSource(dataSource, WatermarkStrategy.noWatermarks(), "dataGenerator")
                .sinkTo(FileSink.<Long>forRowFormat(new Path("output"),
                                new SimpleStringEncoder<>(StandardCharsets.UTF_8.name()))
                        // 添加配置
                        .withOutputFileConfig(OutputFileConfig.builder()
                                .withPartPrefix("passnight-")
                                .withPartSuffix(".log")
                                .build())
                        // 根据日期文件分桶
                        .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd HH", ZoneId.systemDefault()))
                        // 设置滚动策略
                        .withRollingPolicy(DefaultRollingPolicy.builder()
                                .withRolloverInterval(Duration.ofSeconds(10)) // 每10s一个滚动
                                .withMaxPartSize(new MemorySize(1024)) // 每1Kb一个滚动
                                .build())
                        .build());
        env.execute();
    }
}

可以看到数据已经成功输出到文件系统

输出到Kafka

注意假设要配置精确一次, 要开启CheckPoint/事务时间及事务前缀

class KafkaSinkOutput {
    public static void main(String[] args) throws Exception {
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE); // 精确一次必须开启check pint

        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                // 指定kafka地址
                .setBootstrapServers("server.passnight.local:20015,replica.passnight.local:20015,follower.passnight.local:20015")
                .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
                        .setTopic("test") // 设置主题
                        .setValueSerializationSchema(new SimpleStringSchema()) // 设置序列化其
                        .build())
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // 配置"精确一次"
                .setTransactionalIdPrefix("passnight-") // 若要配置精确一次, 必须设置事务
                // 若要配置精确一次, 必须配置事务超时时间; 这个值必须小于transaction.max.timeout.ms=15min
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, String.valueOf(10 * 60 * 1000))
                .build();
        env.socketTextStream("server.passnight.local", 30000)
                .sinkTo(kafkaSink);
        env.execute();
    }
}

可以看到数据已经成功被Kafka 消费者消费

输出到jdbc

先在数据库创建对应的表

CREATE TABLE flink_output
(
    id    INT PRIMARY KEY AUTO_INCREMENT,
    value VARCHAR(32)
)

之后编写java代码操作flink

class JdbcSinkOutput {
    public static void main(String[] args) throws Exception {
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        SinkFunction<String> jdbcSinkFunction = JdbcSink.sink("INSERT INTO flink_output (value) VALUES (?);",
                (JdbcStatementBuilder<String>) (preparedStatement, s) -> preparedStatement.setString(1, s),
                JdbcExecutionOptions.builder()
                        .withMaxRetries(3) // 最多重试3次
                        .withBatchSize(100) // 达到100条一批
                        .withBatchIntervalMs(3000) // **或**3s一批
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost/test")
                        .withUsername("test")
                        .withPassword("123456")
                        .withConnectionCheckTimeoutSeconds(60) // mysql连接会超时, 因此需要心跳保活(默认是8h)
                        .build());
        env.socketTextStream("server.passnight.local", 30000)
                .addSink(jdbcSinkFunction);
        env.execute();
    }
}

使用netcat发送数据后, 可以看到已经有数据被存储到MySQL中了

自定义输出流

对于flink来说, 一般无需使用自定义的输出流, 因为流式处理系统的错误恢复/事务实现非常复杂, 且一般不宜从头开始; 以下是一个可以将数据输出到标准输出的简单例子

class CustomerSinkOutput {
    public static void main(String[] args) throws Exception {
        SinkFunction<String> mySinkFunction = new RichSinkFunction<>() {

            PrintStream output;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                // 创建输出流
                output = System.out;
            }

            /**
             * 调用这个方法输出到标准输出流
             * @param value The input record.
             * @param context Additional context about the input record.
             * @throws Exception 继承父类异常
             */
            @Override
            public void invoke(String value, Context context) throws Exception {
                super.invoke(value, context);
                output.printf("MySink: %s%n", value);
            }

            @Override
            public void close() throws Exception {
                super.close();
                // 关闭输出流
                output = null;
            }
        };
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        env.socketTextStream("server.passnight.local", 30000)
                .addSink(mySinkFunction);
        env.execute();
    }
}

标准输出打印了

MySink: 123132

窗口

1. Flink是一个流式计算引擎, 面对的数据对象是无界数据流; 为了更好地管理和处理该无界流, 可以将无限的数据切割成有限的数据块进行处理, 这个数据块就是窗口

基本概念

分类

1. 按照驱动类型划分

   1. 事件窗口(IMEI Window: 时间窗口是以时间点来定义开始和节数

   2. 计数窗口(Count Window): 基于元素的个数获取数据, 当达到设定的个数时就触发计算并关闭窗口

2. 按照窗口分配数据的规则划分

   1. 滚动窗口(Tumbling Window): 是一种对数据的均匀切片的划分方式, 窗口之间没有重叠 例如每一个小时划分一个窗口就是一个典型的滚动窗口

      

   2. 滑动窗口: 滑动窗口的大小是固定的, 但是窗口之间不是首尾相接而是重叠的 如下图

      1. 滑动窗口有两个参数: 窗口大小和滑动步长
      2. 这样数据可能属于多个窗口
      3. 滚动窗口可以看做特殊的滑动窗口 窗口大小$\le$滑动步长
      4. 

   3. 会话窗口: 会话窗口是基于会话对数据进行分组的, 会话窗口只能基于时间来定义

      1. 会话的特点有: 长度不固定, 事件不固定, 不会重叠
      2. 

   4. 全局窗口: 即将所有的数据都分配到一个窗口, 这种窗口没有结束的时候因此也不会触发计算

窗口分配器

1. 窗口的操作主要分为两个部分: 窗口分配器和窗口函数
2. 窗口分配器指定窗口类型
   1. 窗口分配器除了上一节介绍的分类方式以外, 还有是否有keyBy的窗口, 没有keyBy的窗口所有数据都进入到一个子任务, 且并行度为1
3. 以下是各种类型的窗口声明

    dataSource.keyBy(t -> t).window(TumblingProcessingTimeWindows.of(Time.seconds(10))); // 窗口长度为10的滚动窗口
    dataSource.keyBy(t -> t).window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(2))); // 窗口长度为10, 步长为2的滑动窗口
    dataSource.keyBy(t -> t).window(ProcessingTimeSessionWindows.withGap(Time.seconds(5))); // 超时时长为5s的会话窗口
    dataSource.keyBy(t -> t).countWindow(5); // 窗口长度为5的滚动计数窗口
    dataSource.keyBy(t -> t).countWindow(5, 2); // 窗口长度为5, 滑动步长为2的滑动窗口
    dataSource.keyBy(t -> t).window(GlobalWindows.create()); // 全局窗口(计数窗口的底层就是这个)

窗口函数

1. 窗口函数指定对数据的计算逻辑

reduce函数

class Reduce {
        public static void main(String[] args) throws Exception {
            @Cleanup
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
            env.socketTextStream("server.passnight.local", 30000)
                    .keyBy(t -> t)
                    .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                    .reduce((ReduceFunction<String>) (value1, value2) -> value1 + "|" + value2)
                    .print();
            env.execute();
        }
    }

对于输入:

passnight@passnight-s600:~$ netcat -lk 30000
# 前3s输入
1
1
1
# 后3s输入
2
2
23

输出结果为:

# 第1个3s输出
9> 1|1|1
# 第2个3s输出
5> 2|2
10> 23

可以看到每3s对所有的输入进行keyBy后reduce

aggreage函数

reduce的输入/中间结果/输出的数据类型必须是一致的, 为aggregate的三个类型都可以不一样; 下面是一个统计单词词频的例子

static void main(String[] args) throws Exception {
            @Cleanup
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
            env.socketTextStream("server.passnight.local", 30000)
                    .keyBy(t -> t)
                    .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                    // 第1个类型: 输入数据类型, 代表单词
                    // 第2个类型: 累加器类型(即中间结果类型); 这里代表词频
                    // 第3个类型: 输出类型, 这里代表格式化后的词频表示
                    .aggregate(new AggregateFunction<String, Integer, String>() {

                        @Override
                        public Integer createAccumulator() {
                            System.out.println("WindowAggregate.createAccumulator");
                            return 0;
                        }

                        @Override
                        public Integer add(String value, Integer accumulator) {
                            System.out.println("WindowAggregate.add");
                            return accumulator + 1;
                        }

                        @Override
                        public String getResult(Integer accumulator) {
                            System.out.println("WindowAggregate.getResult");
                            return String.format("词频: %d", accumulator);
                        }

                        // 这个一般只有会话窗口会用到
                        @Override
                        public Integer merge(Integer a, Integer b) {
                            System.out.println("WindowAggregate.merge");
                            return null;
                        }
                    })
                    .print();
            env.execute();
        }

输入为:

1
1
2
34
5
6
6
64
2
4
2
# 第1个3s
3
3
3
3
3
3
3
# 第2个3s
3
3
3
3
3
# 第3个3s
3

输出为:

WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.getResult
9> 词频: 1
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.getResult
WindowAggregate.getResult
9> 词频: 1
15> 词频: 1
WindowAggregate.getResult
13> 词频: 1
WindowAggregate.getResult
5> 词频: 1
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.add
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.getResult
WindowAggregate.getResult
10> 词频: 1
3> 词频: 2
WindowAggregate.getResult
5> 词频: 1
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.createAccumulator
WindowAggregate.add
5> 词频: 1
1> 词频: 1
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.add
WindowAggregate.add
WindowAggregate.add
WindowAggregate.add
WindowAggregate.add
WindowAggregate.add
WindowAggregate.getResult
6> 词频: 7
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.add
WindowAggregate.add
WindowAggregate.add
WindowAggregate.add
WindowAggregate.getResult
6> 词频: 5
WindowAggregate.createAccumulator
WindowAggregate.add
WindowAggregate.add
WindowAggregate.getResult
6> 词频: 1

可以看到每3s调用一个createAccumulate, 并重置计数器

全窗口(process)函数

全窗口函数可以在窗口执行的时候运行, 并且获得全窗口对象等上下文信息

class Process {
        public static void main(String[] args) throws Exception {
            @Cleanup
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
            env.socketTextStream("server.passnight.local", 30000)
                    .keyBy(t -> t)
                    .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                    .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
                        /**
                         * @param s The key for which this window is evaluated.
                         * @param context The context in which the window is being evaluated.
                         * @param elements The elements in the window being evaluated.
                         * @param out A collector for emitting elements.
                         */
                        @Override
                        public void process(String s, ProcessWindowFunction<String, String, String, TimeWindow>.Context context, Iterable<String> elements, Collector<String> out) {
                            String startTimestamp = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss.SSS");
                            String endTimestamp = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss.SSS");
                            long count = elements.spliterator().estimateSize();

                            out.collect(String.format("startTimestamp: %s, endTimestamp: %s, count: %d", startTimestamp, endTimestamp, count));
                        }
                    })
                    .print();
            env.execute();
        }
    }

输入为:

# 第1个3s
1
# 第2个3s
1
1
1
1
1
1

输出为:

9> startTimestamp: 2023-11-12 15:06:03.000, endTimestamp: 2023-11-12 15:06:06.000, count: 1
9> startTimestamp: 2023-11-12 15:06:18.000, endTimestamp: 2023-11-12 15:06:21.000, count: 5

不同窗口类型

时间窗口

    static class AllTimeWindow {
        public static void main(String[] args) throws Exception {
            @Cleanup
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
            env.socketTextStream("server.passnight.local", 30000)
                    .keyBy(t -> t)
//                    .window(TumblingProcessingTimeWindows.of(Time.seconds(3))) // 滚动窗口
//                    .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))) // 长度10s, 步长5s的滑动窗口
//                    .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5))) // 间隔5s的会话窗口
                    .window(ProcessingTimeSessionWindows.withDynamicGap((SessionWindowTimeGapExtractor<String>) element -> Integer.parseInt(element) * 1000L)) // 动态调整间隔时间的会话窗口
                    .reduce((ReduceFunction<String>) (value1, value2) -> value1 + "|" + value2, new ProcessWindowFunction<String, String, String, TimeWindow>() {
                        @Override
                        public void process(String s, ProcessWindowFunction<String, String, String, TimeWindow>.Context context, Iterable<String> elements, Collector<String> out) {
                            String startTimestamp = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss.SSS");
                            String endTimestamp = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss.SSS");
                            long count = elements.spliterator().estimateSize();

                            out.collect(String.format("startTimestamp: %s, endTimestamp: %s, count: %d, data: %s", startTimestamp, endTimestamp, count, elements));
                        }
                    })
                    .print();
            env.execute();
        }
    }

滑动时间窗口

可以看到一个数据被消费了两次, 因为窗口长度是步长的2倍

5> startTimestamp: 2023-11-12 15:24:15.000, endTimestamp: 2023-11-12 15:24:25.000, count: 1, data: [2|2|2|2]
6> startTimestamp: 2023-11-12 15:24:15.000, endTimestamp: 2023-11-12 15:24:25.000, count: 1, data: [3|3|3]
9> startTimestamp: 2023-11-12 15:24:15.000, endTimestamp: 2023-11-12 15:24:25.000, count: 1, data: [1|1|1]
1> startTimestamp: 2023-11-12 15:24:20.000, endTimestamp: 2023-11-12 15:24:30.000, count: 1, data: [4|4|4]
18> startTimestamp: 2023-11-12 15:24:20.000, endTimestamp: 2023-11-12 15:24:30.000, count: 1, data: [7]
6> startTimestamp: 2023-11-12 15:24:20.000, endTimestamp: 2023-11-12 15:24:30.000, count: 1, data: [3|3|3|3]
13> startTimestamp: 2023-11-12 15:24:20.000, endTimestamp: 2023-11-12 15:24:30.000, count: 1, data: [5|5]
9> startTimestamp: 2023-11-12 15:24:20.000, endTimestamp: 2023-11-12 15:24:30.000, count: 1, data: [1|1|1]
5> startTimestamp: 2023-11-12 15:24:20.000, endTimestamp: 2023-11-12 15:24:30.000, count: 1, data: [2|2|2|2]
3> startTimestamp: 2023-11-12 15:24:20.000, endTimestamp: 2023-11-12 15:24:30.000, count: 1, data: [6|6]
9> startTimestamp: 2023-11-12 15:24:25.000, endTimestamp: 2023-11-12 15:24:35.000, count: 1, data: [1]
13> startTimestamp: 2023-11-12 15:24:25.000, endTimestamp: 2023-11-12 15:24:35.000, count: 1, data: [5|5]
3> startTimestamp: 2023-11-12 15:24:25.000, endTimestamp: 2023-11-12 15:24:35.000, count: 1, data: [6|6]
1> startTimestamp: 2023-11-12 15:24:25.000, endTimestamp: 2023-11-12 15:24:35.000, count: 1, data: [4|4|4|4]
18> startTimestamp: 2023-11-12 15:24:25.000, endTimestamp: 2023-11-12 15:24:35.000, count: 1, data: [7]
6> startTimestamp: 2023-11-12 15:24:25.000, endTimestamp: 2023-11-12 15:24:35.000, count: 1, data: [3|3]
6> startTimestamp: 2023-11-12 15:24:30.000, endTimestamp: 2023-11-12 15:24:40.000, count: 1, data: [3]
9> startTimestamp: 2023-11-12 15:24:30.000, endTimestamp: 2023-11-12 15:24:40.000, count: 1, data: [1]
1> startTimestamp: 2023-11-12 15:24:30.000, endTimestamp: 2023-11-12 15:24:40.000, count: 1, data: [4]
13> startTimestamp: 2023-11-12 15:24:30.000, endTimestamp: 2023-11-12 15:24:40.000, count: 1, data: [5]

会话窗口

会话窗口的输出与输入相关, 当5s没有接受到新的输入, 窗口调用一次处理函数

13> startTimestamp: 2023-11-12 15:28:00.116, endTimestamp: 2023-11-12 15:28:12.426, count: 1, data: [5|5|5|5|5|5|5|5|5|5|5]
13> startTimestamp: 2023-11-12 15:28:13.835, endTimestamp: 2023-11-12 15:28:20.638, count: 1, data: [5|5|5|5]
13> startTimestamp: 2023-11-12 15:28:33.563, endTimestamp: 2023-11-12 15:28:41.769, count: 1, data: [5|5|5|5|5|5|5|5|5|5|5]

动态间隔的会话窗口

动态间隔的会话窗口可以根据输入调整间隔时间, 在该例子中, 间隔时间是输入的值Integer.parseInt(element) * 1000L 默认单位是毫秒

6> startTimestamp: 2023-11-12 15:32:48.005, endTimestamp: 2023-11-12 15:32:53.409, count: 1, data: [3|3|3|3|3|3|3|3|3|3|3]
6> startTimestamp: 2023-11-12 15:32:57.121, endTimestamp: 2023-11-12 15:33:01.524, count: 1, data: [3|3|3|3|3|3|3]
13> startTimestamp: 2023-11-12 15:33:20.658, endTimestamp: 2023-11-12 15:33:27.261, count: 1, data: [5|5|5|5|5|5|5|5]
13> startTimestamp: 2023-11-12 15:33:42.489, endTimestamp: 2023-11-12 15:33:55.503, count: 1, data: [5|5|5|5|5|5|5|5|5|5|5|5|5|5|5|5|5]
# 这个输出的时间间隔是8s, 因为flink会按照最新的输入来调整会话时间间隔
3> startTimestamp: 2023-11-12 15:35:39.736, endTimestamp: 2023-11-12 15:35:45.736, count: 1, data: [6]
18> startTimestamp: 2023-11-12 15:35:40.136, endTimestamp: 2023-11-12 15:35:47.136, count: 1, data: [7]
11> startTimestamp: 2023-11-12 15:35:35.350, endTimestamp: 2023-11-12 15:35:50.340, count: 1, data: [8|8|8|8|8|8|8|8]

计数窗口

    static class AllCountWindow {
        public static void main(String[] args) throws Exception {
            @Cleanup
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
            env.socketTextStream("server.passnight.local", 30000)
                    .keyBy(t -> t)
                    .countWindow(5)
                    .reduce((ReduceFunction<String>) (value1, value2) -> value1 + "|" + value2, new ProcessWindowFunction<String, String, String, GlobalWindow>() {
                        @Override
                        public void process(String s, ProcessWindowFunction<String, String, String, GlobalWindow>.Context context, Iterable<String> elements, Collector<String> out) {
                            String currentTimestamp = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
                            long count = elements.spliterator().estimateSize();

                            out.collect(String.format("currentTimestamp: %s, count: %d, data: %s", currentTimestamp, count, elements));
                        }
                    })
                    .print();
            env.execute();
        }
    }
}

简单滑动窗口

计数窗口的执行和时间没有关系, 只和数量有关系, 下面每有5个元素就完成一次输出

9> currentTimestamp: 2023-11-12 15:44:56.391, count: 1, data: [1|1|1|1|1]
9> currentTimestamp: 2023-11-12 15:44:59.464, count: 1, data: [1|1|1|1|1]
9> currentTimestamp: 2023-11-12 15:45:00.566, count: 1, data: [1|1|1|1|1]
9> currentTimestamp: 2023-11-12 15:45:03.771, count: 1, data: [1|1|1|1|1]

带步长的滑动窗口

滑动窗口每经过一个步长会触发一次, 下面蓝色的代表着滑动窗口

时间语义

Flink中的时间大致可以分为两类; 而具体使用哪种时间作为衡量标准, 就是时间语义:

1. 事件事件: 数据产生的时间
2. 处理时间: 数据开始被处理的时间

水位线

1. 在Flink中, 用来衡量事件事件进展的标记, 被称为水位线

有序流中的水位线

1. 理想状态下:
   1. 数据按照生成的先后顺序进入流中, 每条数据产生一个水位线
   2. 

实际状态下的水位线:

1. 如果当前&数据量非常大, 且同时涌来的数据时间差非常小, 其对处理计算没有什么影响, 为了提高效率, 每隔一段时间生成一个水位线:

2. 乱序流中的水位线

   1. 在分布式系统中, 数据在节点间传输, 可能会因为网络延迟的不确定性, 导致数据顺序发生改变, 这就是所谓的乱序数据
   2. 

3. 乱序+数据量小: 为数据添加一个时间戳, 当新的数据到达时, 倘若时间戳是否比之前大, 若比之前小则不生成新的水位线 即只有更大的时间戳到达时才生成新的水位线:

4. 乱序+数据量大: 因为数据量过大, 只能周期性地生成水位线, 即取当前时间间隔到达的的最大时间戳为水位线:

5. 乱序+数据量大+迟到数据: 为了让窗口能够接收到迟到数据, 可以等待一段时间, 即将当前间隔时间最大的时间戳减去几秒作为水位线

   1. 如下图, W(9)发生了数据迟到:

   2. 为了能够包括迟到的8和9, 可以将前四个元素的水位线设置为$w(9-2)$, 这样当8和9到达时, 也不会被丢弃 触发窗口会触发数据执行及窗口关闭

   3. 

6. 水位线特性:

   1. 水位线时插入到数据流中的一个标记, 可以认为是特殊的数据
   2. 水位线主要的内容是一个时间戳, 用来表示当前事件时间的进展
   3. 水位线是基于数据的时间戳生成的
   4. 水位线的时间戳必须单调递增, 以确保任务的事件时间始终一直向前推进
   5. 水位线可以通过设置延迟来保证正确处理乱序数据
   6. 一个水位线, 刚表示当前流中的事件时间已经达到了时间戳t, 这代表t之前的所有数据都到齐了, 之后流中就不会出现时间戳小于水位线的数据 这个等待机制是flink实现对乱序数据处理的方法

水位线和窗口的关系

1. 误解: flink中窗口是用来处理无界流的核心, 数据源源源不断地流过来, 到某个时间点窗口该关闭了, 就停止收集数据并触发计算及数据结果 窗口是包含水位线等待的数据的
2. 窗口是多个存储桶, 数据会被发送到对应的存储桶当中 ,当到达窗口的时间结束时, 就对每个桶汇总的数据进行计算处理 如下图, 等待时间接收到超前的数据应该放到下一个水位线的桶当中:

水位线生成

1. 生成原则:
   1. 保证能够处理的数据是正确的, 到齐的, 之后不会在出现该事件区间的数据
   2. 但实际情况下, 为了保证绝对正确, 就必须等待足够长的数据, 这样会带来更高的延迟 因此正确性和性能是一对互斥的需求
2. 生成策略: flink当中, 可以对流调用assignTimestampsAndWatermarks()来为数据分配时间戳

有序流水位线

class OrderedWatermark {
    public static void main(String[] args) throws Exception {
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.socketTextStream("server.passnight.local", 30000)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<String>forMonotonousTimestamps()
                        /* 指定时间戳提取器, 可以从element中提取参数, 也可以直接定义为recordTimestamp
                         * @param element The element that the timestamp will be assigned to.
                         * @param recordTimestamp The current internal timestamp of the element, or a negative value, if
                         *  no timestamp has been assigned yet.
                         */
                        .withTimestampAssigner((SerializableTimestampAssigner<String>) (element, recordTimestamp) -> Integer.parseInt(element.split(" ")[1]) * 1000L))
                .keyBy(t -> t)
                // 要使用Watermark, 需要指定事件时间的窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(3)))
                .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<String, String, String, TimeWindow>.Context context, Iterable<String> elements, Collector<String> out) {
                        String startTimestamp = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss.SSS");
                        String endTimestamp = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss.SSS");
                        long count = elements.spliterator().estimateSize();

                        out.collect(String.format("startTimestamp: %s, endTimestamp: %s, count: %d, elements: %s", startTimestamp, endTimestamp, count, elements));
                    }
                })
                .print();
        env.execute();
    }
}

输入为:

1 1
1 1
1 2
1 3

因为时间被定义为第二个数字秒, 所以只有当1 4输入时, 才会产生关闭窗口, 触发计算; 且

10> startTimestamp: 1970-01-01 08:00:00.000, endTimestamp: 1970-01-01 08:00:03.000, count: 2, elements: [1 1, 1 1]
1> startTimestamp: 1970-01-01 08:00:00.000, endTimestamp: 1970-01-01 08:00:03.000, count: 1, elements: [1 2]

乱序流水位线

class UnorderedWatermark {
    public static void main(String[] args) throws Exception {
        @Cleanup
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.socketTextStream("server.passnight.local", 30000)
                // 等待3秒时间
                .assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((SerializableTimestampAssigner<String>) (element, recordTimestamp) -> Integer.parseInt(element.split(" ")[1]) * 1000L))
                .keyBy(t -> t)
                // 要使用Watermark, 需要指定事件时间的窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(3)))
                .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<String, String, String, TimeWindow>.Context context, Iterable<String> elements, Collector<String> out) {
                        String startTimestamp = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss.SSS");
                        String endTimestamp = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss.SSS");
                        long count = elements.spliterator().estimateSize();

                        out.collect(String.format("startTimestamp: %s, endTimestamp: %s, count: %d, elements: %s", startTimestamp, endTimestamp, count, elements));
                    }
                })
                .print();
        env.execute();
    }
}