认识Flink

Docker安装Flink

version: "2.1"
services:jobmanager:image: flinkexpose:- "6123"ports:- "20010:8081"command: jobmanagerenvironment:- JOB_MANAGER_RPC_ADDRESS=jobmanagertaskmanager:image: flinkexpose:- "6121"- "6122"depends_on:- jobmanagercommand: taskmanagerlinks:- "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.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 {@CleanupStreamExecutionEnvironment 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 {@CleanupStreamExecutionEnvironment 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), // 生成器方法, 主要重写里面的map10, // 自动生成的序列号的最大值, 达到这个值之后就停止生成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";@Overridepublic 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 {@CleanupStreamExecutionEnvironment 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>() {@Overridepublic Integer map(WaterSensor value) throws Exception {return value.getVc();}// 富函数带的钩子@Overridepublic void open(Configuration parameters) {RuntimeContext context = getRuntimeContext();System.out.printf("[open()] 子任务编号: %s, 子任务名称: %s%n",context.getIndexOfThisSubtask(),context.getTaskNameWithSubtasks());}@Overridepublic 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));@CleanupStreamExecutionEnvironment 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<>() {@Overridepublic 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 {@CleanupStreamExecutionEnvironment 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 {@CleanupStreamExecutionEnvironment 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>() {@Overridepublic String map1(Integer value) {return String.valueOf(value * 10);}@Overridepublic 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 {@CleanupStreamExecutionEnvironment 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 {@CleanupStreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE); // 精确一次必须开启check pintKafkaSink<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 {@CleanupStreamExecutionEnvironment 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;@Overridepublic 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 继承父类异常*/@Overridepublic void invoke(String value, Context context) throws Exception {super.invoke(value, context);output.printf("MySink: %s%n", value);}@Overridepublic void close() throws Exception {super.close();// 关闭输出流output = null;}};@CleanupStreamExecutionEnvironment 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 {@CleanupStreamExecutionEnvironment 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 {@CleanupStreamExecutionEnvironment 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>() {@Overridepublic Integer createAccumulator() {System.out.println("WindowAggregate.createAccumulator");return 0;}@Overridepublic Integer add(String value, Integer accumulator) {System.out.println("WindowAggregate.add");return accumulator + 1;}@Overridepublic String getResult(Integer accumulator) {System.out.println("WindowAggregate.getResult");return String.format("词频: %d", accumulator);}// 这个一般只有会话窗口会用到@Overridepublic 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 {@CleanupStreamExecutionEnvironment 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.*/@Overridepublic 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 {@CleanupStreamExecutionEnvironment 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>() {@Overridepublic 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 {@CleanupStreamExecutionEnvironment 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>() {@Overridepublic 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 {@CleanupStreamExecutionEnvironment 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>() {@Overridepublic 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();}
}