桥下の民 - Flink
type: channel
link: https://www.wordplay.work/index.php/category/Flink
items: Dinky
items: Flink-CDC
items: Flink CEP
items: Flink Table API 与 SQL
items: Flink 状态编程和容错机制
items: Flink ProcessFunction API（底层 API）
items: Flink 时间语义与 Wartermark
items: Flink 中的 Window
items: Flink 流处理API
items: Flink 部署&运行架构

Dinky
type: item
date: 2024-01-26T01:26:00+00:00
description: 一、概述 1.1、简介 Dinky 是一个开箱即用、易扩展，以 Apache Flink 为基础，连接 OLAP 和数据湖等众多框架的一站式实时计算平台，致力于流批一体和湖仓一体的探索与实践。 官网：http://www.dlink.top/ 1.2、核心特性 1）沉浸式 提供专业的 DataStudio 功能，支持全屏开发、自动提示与补全、语法高亮、语句美化、语法校验、 调试预览结果、全局变量、MetaStore、字段级血缘分析、元数据查询、FlinkSQL 生成等功能。 2）易用性 Flink 多种执行模式无感知切换，支持 Flink 多版本切换，自动化托管实时任务、恢复点、报警等， 自定义各种配置，持久化管理的 Flink Catalog。 3）增强式 兼容且增强官方 FlinkSQL 语法，如 SQL 表值聚合函数、全局变量、CDC 整库同步、执行环境、 语句合并、共享会话等。 4）一站式 提供从 FlinkSQL 开发调试到上线下线的运维监控及 SQL 的查询执行能力，使数仓建设及数据治理一体化。 5）易扩展 源码采用 SPI 插件化及各种设计模式支持用户快速扩展新功能，如连接器、数据源、报警方式、 Flink Catalog、CDC 整库同步、自定义 FlinkSQL 语法等。 6）无侵入 Spring Boot 轻应用快速部署，不需要在任何 Flink 集群修改源码或添加额外插件，无感知连接和监控Flink 集群。 1.3、功能 沉浸式 FlinkSQL 数据开发：自动提示补全、语法高亮、语句美化、在线调试、语法校验、执行计划、MetaStore、血缘分析、版本对比等。 支持 FlinkSQL 多版本开发及多种执行模式：Local、Standalone、Yarn/Kubernetes Session、Yarn Per-Job、Yarn/Kubernetes Application。 支持 Apache Flink 生态：Connector、FlinkCDC、Table Store 等。 支持 FlinkSQL 语法增强：表值聚合函数、全局变量、执行环境、语句合并、整库同步等。 支持 FlinkCDC 整库实时入仓入湖、多库输出、自动建表、模式演变。 支持 Flink Java / Scala / Python UDF 开发与自动提交。 支持 SQL 作业开发：ClickHouse、Doris、Hive、Mysql、Oracle、Phoenix、PostgreSql、Presto、SqlServer、StarRocks 等。 支持实时在线调试预览 Table、 ChangeLog、统计图和 UDF。 支持 Flink Catalog、数据源元数据在线查询及管理。 支持自动托管的 SavePoint/CheckPoint 恢复及触发机制：最近一次、最早一次、指定一次等。 支持实时任务运维：上线下线、作业信息、集群信息、作业快照、异常信息、数据地图、数据探查、历史版本、报警记录等。 支持作为多版本 FlinkSQL Server 以及 OpenApi 的能力。 支持实时作业报警及报警组：钉钉、微信企业号、飞书、邮箱等。 支持多种资源管理：集群实例、集群配置、Jar、数据源、报警组、报警实例、文档、系统配置等。 支持企业级管理功能：多租户、用户、角色、命名空间等。 二、安装部署 Dinky 不依赖任何外部的 Hadoop 或者 Flink 环境，可以单独部署在 flink、 hadoop 和 K8S 集群之外，完全解耦，支持同时连接多个不同的集群实例进行运维。 2.1、解压到指定目录 上传安装包并解压： tar -zxvf dlink-release-0.7.3.tar.gz -C /opt/module/ mv dlink-release-0.7.3 dinky cd dinky 2.2、初始化MySQL数据库 Dinky 采用 mysql 作为后端的存储库，部署需要 MySQL5.7 以上版本，这里我们使用的 MySQL 是 8.0。 在 Dinky 根目录 sql 文件夹下分别放置了初始化的dinky.sql文件、升级使用的upgrade/${version}_schema/mysql/ddl 和 dml。如果第一次部署，可以直接将 sql/dinky.sql 文件在 dinky 数据库下执行。（如果之前已经部署，那 upgrade 目录下存放了各版本的升级 sql ，根据版本号按需执行即可）。 在MySQL中操作如下： #创建数据库 mysql> CREATE DATABASE dinky; #创建用户并允许远程登录 mysql> create user 'dinky'@'%' IDENTIFIED WITH m…

Flink-CDC
type: item
date: 2023-01-22T07:35:00+00:00
description: 一、CDC 简介 1.1、CDC CDC 是 Change Data Capture(变更数据获取) 的简称。核心思想是，监测并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。 1.2、CDC 的种类 CDC 主要分为基于查询和基于 Binlog 两种方式，主要了解一下这两种之间的区别： 1.3、Flink-CDC Flink 社区开发了 flink-cdc-connectors 组件，这是一个可以直接从 MySQL、PostgreSQL 等数据库直接读取全量数据和增量变更数据的 source 组件。 目前也已开源，开源地址：https://github.com/ververica/flink-cdc-connectors 二、FlinkCDC 示例 2.1、数据准备 2.1.1、在MySQL中创建数据库及表 2.1.2、插入数据 1）在 test 数据库中插入数据 2）在 test_route 数据库中插入数据 2.1.3、开启 MySQL Binlog 并重启 MySQL [atguigu@hadoop103 ~]$ sudo vim /etc/my.cnf #添加如下配置信息,开启 test 以及 test_route 数据库的 Binlog #数据库id server-id = 1 ##启动binlog，该参数的值会作为binlog的文件名 log-bin=mysql-bin ##binlog类型，maxwell要求为row类型 binlog_format=row ##启用binlog的数据库，需根据实际情况作出修改 binlog-do-db=test binlog-do-db=test_route 2.2、DataStream方式的应用 2.2.1、导入依赖 2.2.2、编写代码 详见代码。 2.2.3、案例测试 1）打包并上传至 Linux 2）启动 HDFS 集群 [atguigu@hadoop102 flink-1.18.0]$ start-dfs.sh 3）启动 Flink 集群 [atguigu@hadoop102 flink-1.18.0]$ bin/start-cluster.sh 4）启动程序 [atguigu@hadoop102 flink-1.18.0]$ bin/flink run -m hadoop102:8081 -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar 5）观察 TaskManager 日志，会从头读取表数据 6）给当前的 Flink 程序创建 Savepoint [atguigu@hadoop102 flink-local]$ bin/flink savepoint JobId hdfs://hadoop102:8020/flinkCDC/save 在 WebUI 中 cancelJob 。 在 MySQL 的 test.t1 表中添加、修改或者删除数据。 从 Savepoint 重启程序。 [atguigu@hadoop102 flink-standalone]$ bin/flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-5dadae-02c69ee54885 -c com.atguigu.cdc. FlinkCDCDataStreamTest ./gmall-flink-cdc.jar 2.3、FlinkSQL 方式的应用 2.3.1、编写代码 详见代码 2.3.2、代码测试 直接运行即可，打包到集群测试与 DataStream 相同！ 2.4、MySQL 到 Doris 的 StreamingETL 实现（3.0） 2.4.1、环境准备 1）安装 FlinkCDC [atguigu@hadoop102 software]$ tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ 2）拖入 MySQL 以及 Doris 依赖包 将flink-cdc-pipeline-connector-doris-3.0.0.jar以及flink-cdc-pipeline-connector-mysql-3.0.0.jar防止在FlinkCDC 的 lib 目录下。 2.4.2、同步变更 2）编写 MySQL 到 Doris 的同步变更配置文件 在 FlinkCDC 目录下创建 job 文件夹，并在该目录下创建同步变更配置文件。 [atguigu@hadoop102 flink-cdc-3.0.0]$ mkd…

Flink CEP
type: item
date: 2022-03-15T15:58:00+00:00
description: 一、什么是复杂事件处理 CEP 一个或多个由简单事件构成的事件流通过一定的规则匹配，然后输出用户想得到的数据，满足规则的复杂事件。 特征： 目标：从有序的简单事件流中发现一些高阶特征。 输入：一个或多个由简单事件构成的事件流。 处理：识别简单事件之间的内在联系，多个符合一定规则的简单事件构成复杂事件。 输出：满足规则的复杂事件。 CEP 用于分析低延迟、频繁产生的不同来源的事件流。CEP 可以帮助在复杂的、不相关的事件流中找出有意义的模式和复杂的关系，以接近实时或准实时的获得通知并阻止一些行为。 CEP 支持在流上进行模式匹配，根据模式的条件不同，分为连续的条件或不连续的条件；模式的条件允许有时间的限制，当在条件范围内没有达到满足的条件时，会导致模式匹配超时。 看起来很简单，但是它有很多不同的功能： 输入的流数据，尽快产生结果。 在 2 个 event 流上，基于时间进行聚合类的计算。 提供实时/准实时的警告和通知。 在多样的数据源中产生关联并分析模式。 高吞吐、低延迟的处理。 市场上有多种 CEP 的解决方案，例如 Spark、Samza、Beam 等，但他们都没有提供专门的 library 支持。但是 Flink 提供了专门的 CEP library。 二、Flink CEP Flink 为 CEP 提供了专门的 Flink CEP library，它包含如下组件： Event Stream pattern 定义 pattern 检测 生成 Alert 首先，开发人员要在 DataStream 流上定义出模式条件，之后 Flink CEP 引擎进行模式检测，必要时生成告警。 为了使用 Flink CEP，需要导入依赖： <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-cep_2.12</artifactId> <version>1.10.1</version> </dependency> 2.1、Pattern API 每个 Pattern 都应该包含几个步骤，或者叫做 state。从一个 state 到另一个 state，通常需要定义一些条件，例如下列的代码： val loginFailPattern = Pattern.begin[LoginEvent]("begin") .where(_.eventType.equals("fail")) .next("next") .where(_.eventType.equals("fail")) .within(Time.seconds(10) 每个 state 都应该有一个标示：例如.begin[LoginEvent]("begin")中的"begin"。 每个 state 都需要有一个唯一的名字，而且需要一个 filter 来过滤条件，这个过滤条件定义事件需要符合的条件，例如: .where(_.eventType.equals("fail")) 也可以通过 subtype 来限制 event 的子类型： start.subtype(SubEvent.class).where(...); 事实上，你可以多次调用 subtype 和 where 方法；而且如果 where 条件是不相关的，你可以通过 or 来指定一个单独的 filter 函数： pattern.where(...).or(...); 之后，可以在此条件基础上，通过 next 或者 followedBy 方法切换到下一个 state，next 的意思是说上一步符合条件的元素之后紧挨着的元素；而 followedBy 并不要求一定是挨着的元素。这两者分别称为严格近邻和非严格近邻。 val strictNext = start.next("middle") val nonStrictNext = start.followedBy("middle") 最后，可以将所有的 Pattern 的条件限定在一定的时间范围内： next.within(Time.seconds(10)) 这个时间可以是 Processing Time，也可以是 Event Time。 2.2、Pattern 检测 通过一个 input DataStream 以及刚刚定义的 Pattern，可以创建一个 PatternStream val input = ... val pattern = ... val patternStream = CEP.pattern(input, pattern) val patternStream = CEP.pattern(loginEventStream.keyBy(_.userId), loginFailPattern) 一旦…

Flink Table API 与 SQL
type: item
date: 2022-03-15T15:54:00+00:00
description: Table API 是流处理和批处理通用的关系型 API，Table API 可以基于流输入或者批输入来运行而不需要进行任何修改。Table API 是 SQL 语言的超集并专门为 Apache Flink 设计的，Table API 是 Scala 和 Java 语言集成式的 API。与常规 SQL 语言中将查询指定为字符串不同，Table API 查询是以 Java 或 Scala 中的语言嵌入样式来定义的，具有 IDE 支持如：自动完成和语法检测。 一、依赖 <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>1.10.1</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-scala-bridge_2.12</artifactId> <version>1.10.1</version> </dependency> 二、TableAPI def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) val inputStream = env.readTextFile("..\\sensor.txt") val dataStream = inputStream .map(data => { val dataArray = data.split(",") SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble) } ) // 基于 env 创建 tableEnv val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build() val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings) // 从一条流创建一张表 val dataTable: Table = tableEnv.fromDataStream(dataStream) // 从表里选取特定的数据 val selectedTable: Table = dataTable.select('id, 'temperature) .filter("id = 'sensor_1'") val selectedStream: DataStream[(String, Double)] = selectedTable .toAppendStream[(String, Double)] selectedStream.print() env.execute("table test") } 2.1、动态表 如果流中的数据类型是 case class 可以直接根据 case class 的结构生成 table。 tableEnv.fromDataStream(dataStream) 或者根据字段顺序单独命名。 tableEnv.fromDataStream(dataStream,’id,’timestamp .......) 最后的动态表可以转换为流进行输出。 table.toAppendStream[(String,String)] 2.2、字段 用一个单引放到字段前面来标识字段名, 如 ‘name , ‘id ,’amount 等 三、TableAPI 的窗口聚合操作 3.1、通过一个例子了解 TableAPI // 统计每 10 秒中每个传感器温度值的个数 def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) val inputStream = env.readTextFile("..\\sensor.txt") …

Flink 状态编程和容错机制
type: item
date: 2022-03-15T14:44:00+00:00
description: 流式计算分为无状态和有状态两种情况。无状态的计算观察每个独立事件，并根据最后一个事件输出结果。例如，流处理应用程序从传感器接收温度读数，并在温度超过 90 度时发出警告。有状态的计算则会基于多个事件输出结果。以下是一些例子。 所有类型的窗口。例如，计算过去一小时的平均温度，就是有状态的计算。 所有用于复杂事件处理的状态机。例如，若在一分钟内收到两个相差 20 度以上的温度读数，则发出警告，这是有状态的计算。 流与流之间的所有关联操作，以及流与静态表或动态表之间的关联操作，都是有状态的计算。 下图展示了无状态流处理和有状态流处理的主要区别。无状态流处理分别接收每条数据记录(图中的黑条)，然后根据最新输入的数据生成输出数据(白条)。有状态流处理会维护状态(根据每条输入记录进行更新)，并基于最新输入的记录和当前的状态值生成输出记录(灰条) 上图中输入数据由黑条表示。无状态流处理每次只转换一条输入记录，并且仅根据最新的输入记录输出结果(白条)。有状态流处理维护所有已处理记录的状态值，并根据每条新输入的记录更新状态，因此输出记录(灰条)反映的是综合考虑多个事件之后的结果。 尽管无状态的计算很重要，但是流处理对有状态的计算更感兴趣。事实上，正确地实现有状态的计算比实现无状态的计算难得多。旧的流处理系统并不支持有状态的计算，而新一代的流处理系统则将状态及其正确性视为重中之重。 一、有状态的算子和应用程序 Flink 内置的很多算子，数据源 source，数据存储 sink 都是有状态的，流中的数据都是 buffer records，会保存一定的元素或者元数据。例如: ProcessWindowFunction 会缓存输入流的数据，ProcessFunction 会保存设置的定时器信息等等。 在 Flink 中，状态始终与特定算子相关联。总的来说，有两种类型的状态： 算子状态（operator state） 键控状态（keyed state） 1.1、算子状态（operator state） 算子状态的作用范围限定为算子任务。这意味着由同一并行任务所处理的所有数据都可以访问到相同的状态，状态对于同一任务而言是共享的。算子状态不能由相同或不同算子的另一个任务访问。 Flink 为算子状态提供三种基本数据结构： 列表状态（List state） 将状态表示为一组数据的列表。 联合列表状态（Union list state） 也将状态表示为数据的列表。它与常规列表状态的区别在于，在发生故障时，或者从保存点（savepoint）启动应用程序时如何恢复。 广播状态（Broadcast state） 如果一个算子有多项任务，而它的每项任务状态又都相同，那么这种特殊情况最适合应用广播状态。 1.2、键控状态（keyed state） 键控状态是根据输入数据流中定义的键（key）来维护和访问的。Flink 为每个键值维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中，这个任务会维护和处理这个 key 对应的状态。当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的 key。因此，具有相同 key 的所有数据都会访问相同的状态。Keyed State 很类似于一个分布式的 key-value map 数据结构，只能用于 KeyedStream（keyBy 算子处理之后）。 Flink 的 Keyed State 支持以下数据类型： ValueState<T> 保存单个的值，值的类型为 T。 get 操作: ValueState.value() set 操作: ValueState.update(T value) ListState<T> 保存一个列表，列表里的元素的数据类型为 T。基本操作如下： ListState.add(T value) ListState.addAll(List values) ListState.get()返回 Iterable ListState.update(List values) MapState<K, V> 保存 Key-Value 对。 MapState.get(UK key) MapState.put(UK key, UV value) MapState.contains(UK key) MapState.remove(UK key) ReducingState<T> AggregatingState<I, O> State.clear() 是清空操作。 可以利用 Keyed State，实现这样一个需求：检测传感器的温度值，如果连续的两个温度差值超过 10 度，就输出报警。 DataStream<Tuple3<String, Double, Double>> warningStream = d…

Flink ProcessFunction API（底层 API）
type: item
date: 2022-03-15T14:32:00+00:00
description: 转换算子是无法访问事件的时间戳信息和水位线信息的。而这在一些应用场景下，极为重要。例如 MapFunction 这样的 map 转换算子就无法访问时间戳或者当前事件的事件时间。 基于此，DataStream API 提供了一系列的 Low-Level 转换算子。可以访问时间戳、watermark 以及注册定时事件。还可以输出特定的一些事件，例如超时事件等。Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑(使用之前的 window 函数和转换算子无法实现)。例如，Flink SQL 就是使用 Process Function 实现的。 Flink 提供了 8 个 Process Function： ProcessFunction KeyedProcessFunction CoProcessFunction ProcessJoinFunction BroadcastProcessFunction KeyedBroadcastProcessFunction ProcessWindowFunction ProcessAllWindowFunction 一、KeyedProcessFunction KeyedProcessFunction 用来操作 KeyedStream。KeyedProcessFunction 会处理流的每一个元素，输出为 0 个、1 个或者多个元素。所有的 Process Function 都继承自 RichFunction 接口，所以都有 open()、close()和 getRuntimeContext()等方法。而 KeyedProcessFunction<K, I, O> 还额外提供了两个方法 processElement(I value, Context ctx, Collector<O> out), 流中的每一个元素都会调用这个方法，调用结果将会放在 Collector 数据类型中输出。Context 可以访问元素的时间戳，元素的 key，以及 TimerService 时间服务。Context 还可以将结果输出到别的流(side outputs)。 onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) 是一个回调函数。当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的触发的时间戳。Collector 为输出结果的集合。OnTimerContext 和 processElement 的 Context 参数一样，提供了上下文的一些信息，例如定时器触发的时间信息(事件时间或者处理时间)。 二、TimerService 和定时器（Timers） Context 和 OnTimerContext 所持有的 TimerService 对象拥有以下方法: long currentProcessingTime() 返回当前处理时间 long currentWatermark() 返回当前 watermark 的时间戳 void registerProcessingTimeTimer(long timestamp) 会注册当前 key 的 processing time 的定时器。当 processing time 到达定时时间时，触发 timer。 void registerEventTimeTimer(long timestamp) 会注册当前 key 的 event time 定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。 void deleteProcessingTimeTimer(long timestamp) 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行。 void deleteEventTimeTimer(long timestamp) 删除之前注册的事件时间定时器，如果没有此时间戳的定时器，则不执行。 当定时器 timer 触发时，会执行回调函数 onTimer()。注意定时器 timer 只能在 keyed streams 上面使用。 下面举个例子说明 KeyedProcessFunction 如何操作 KeyedStream。 需求：监控温度传感器的温度值，如果温度值在 10 秒钟之内(processing time)连续上升，则报警。 DataStream<String> warningStream = dataStream.keyBy(SensorReading::getId) .process( new TempIncreaseWarning(10) ); 看一下 TempIncreaseWarning 如何实现, 程序中使用了…

Flink 时间语义与 Wartermark
type: item
date: 2022-03-15T14:02:00+00:00
description: 一、Flink 中的时间语义 在 Flink 的流式处理中，会涉及到时间的不同概念，如下图所示： Event Time：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink 通过时间戳分配器访问事件时间戳。 Ingestion Time：是数据进入 Flink 的时间。 Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是 Processing Time。 例如，一条日志进入 Flink 的时间为 2017-11-12 10:00:00.123，到达 Window 的系统时间为 2017-11-12 10:00:01.234，日志的内容如下： 2017-11-02 18:37:15.624 INFO Fail over to rm2 对于业务来说，要统计 1min 内的故障日志个数，哪个时间是最有意义的？—— eventTime，因为要根据日志的生成时间进行统计。 二、EventTime 的引入 在 Flink 的流式处理中，绝大部分的业务都会使用 eventTime，一般只在 eventTime 无法使用时，才会被迫使用 ProcessingTime 或者 IngestionTime。 如果要使用 EventTime，那么需要引入 EventTime 的时间属性，引入方式如下所示： StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment // 从调用时刻开始给 env 创建的每一个 stream 追加时间特征 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 三、Watermark 3.1、基本概念 流处理从事件产生，到流经 source，再到 operator，中间是有一个过程和时间的，虽然大部分情况下，流到 operator 的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、分布式等原因，导致乱序的产生，所谓乱序，就是指 Flink 接收到的事件的先后顺序不是严格按照事件的 Event Time 顺序排列的。 那么此时出现一个问题，一旦出现乱序，如果只根据 eventTime 决定 window 的运行，我们不能明确数据是否全部到位，但又不能无限期的等下去，此时必须要有个机制来保证一个特定的时间后，必须触发 window 去进行计算了，这个特别的机制，就是 Watermark。 Watermark 是一种衡量 Event Time 进展的机制。 Watermark 是用于处理乱序事件的，而正确的处理乱序事件，通常用Watermark 机制结合 window 来实现。 数据流中的 Watermark 用于表示 timestamp 小于 Watermark 的数据，都已经到达了，因此，window 的执行也是由 Watermark 触发的。 Watermark 可以理解成一个延迟触发机制，我们可以设置 Watermark 的延时时长 t，每次系统会校验已经到达的数据中最大的 maxEventTime，然后认定 eventTime 小于 maxEventTime - t 的所有数据都已经到达，如果有窗口的停止时间等于 maxEventTime – t，那么这个窗口被触发执行。 有序流的 Watermarker 如下图所示：（Watermark 设置为 0） 乱序流的 Watermarker 如下图所示：（Watermark 设置为 2） 当 Flink 接收到数据时，会按照一定的规则去生成 Watermark，这条 Watermark 就等于当前所有到达数据中的 maxEventTime - 延迟时长，也就是说，Watermark 是基于数据携带的时间戳生成的，一旦 Watermark 比当前未触发的窗口的停止时间要晚，那么就会触发相应窗口的执行。由于 event time 是由数据携带的，因此，如果运行过程中无法获取新的数据，那么没有被触发的窗口将永远都不被触发。 上图中，我们设置的允许最大延迟到达时间为 2s，所以时间戳为 7s 的事件对应的 Watermark 是 5s，时间戳为 12s 的事件的 Watermark 是 10s，如果我们的窗口 1是 1s~5s，窗口 2 是 6s~10s，那么时间戳为 7s 的事件到达时的 Watermarker 恰好触发窗口 1，时间戳为 12s 的事件到达时的 Watermark 恰好触发窗口 2。 Watermark 就是触发前一窗口的关窗时间，一旦触发关门那么以当前时刻为…

Flink 中的 Window
type: item
date: 2022-03-15T13:13:00+00:00
description: 一、Window 1.1、Window 概述 streaming 流式计算是一种被设计用于处理无限数据集的数据处理引擎，而无限数据集是指一种不断增长的本质上无限的数据集，而 window 是一种切割无限数据为有限块进行处理的手段。 Window 是无限数据流处理的核心，Window 将一个无限的 stream 拆分成有限大小的 buckets 桶，我们可以在这些桶上做计算操作。 1.2、Window 类型 Window 可以分成两类： CountWindow：按照指定的数据条数生成一个 Window，与时间无关。 TimeWindow：按照时间生成 Window。 对于 TimeWindow，可以根据窗口实现原理的不同分成三类：滚动窗口（Tumbling Window）、滑动窗口（Sliding Window）和会话窗口（Session Window）。 1.2.1、滚动窗口（Tumbling Windows） 将数据依据固定的窗口长度对数据进行切片。 特点：时间对齐，窗口长度固定，没有重叠。 滚动窗口分配器将每个元素分配到一个指定窗口大小的窗口中，滚动窗口有一个固定的大小，并且不会出现重叠。例如：如果你指定了一个 5 分钟大小的滚动窗口，窗口的创建如下图所示： 适用场景：适合做 BI 统计等（做每个时间段的聚合计算）。 1.2.2、滑动窗口（Sliding Windows） 滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成。 特点：时间对齐，窗口长度固定，可以有重叠。 滑动窗口分配器将元素分配到固定长度的窗口中，与滚动窗口类似，窗口的大小由窗口大小参数来配置，另一个窗口滑动参数控制滑动窗口开始的频率。因此，滑动窗口如果滑动参数小于窗口大小的话，窗口是可以重叠的，在这种情况下元素会被分配到多个窗口中。 例如，你有 10 分钟的窗口和 5 分钟的滑动，那么每个窗口中 5 分钟的窗口里包含着上个 10 分钟产生的数据，如下图所示： 适用场景：对最近一个时间段内的统计（求某接口最近 5min 的失败率来决定是否要报警）。 1.2.3、会话窗口（Session Windows） 由一系列事件组合一个指定时间长度的 timeout 间隙组成，类似于 web 应用的bsession，也就是一段时间没有接收到新数据就会生成新的窗口。 特点：时间无对齐。 session 窗口分配器通过 session 活动来对元素进行分组，session 窗口跟滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况，相反，当它在一个固定的时间周期内不再收到元素，即非活动间隔产生，那个这个窗口就会关闭。一个 session 窗口通过一个 session 间隔来配置，这个 session 间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的 session 将关闭并且后续的元素将被分配到新的 session 窗口中去。 二、Window API 2.1、TimeWindow TimeWindow 是将指定时间范围内的所有数据组成一个 window，一次对一个window 里面的所有数据进行计算。 2.1.1、滚动窗口 Flink 默认的时间窗口根据 Processing Time 进行窗口的划分，将 Flink 获取到的数据根据进入 Flink 的时间划分到不同的窗口中。 DataStream<Tuple2<String, Double>> minTempPerWindowStream = dataStream .map(new MapFunction<SensorReading, Tuple2<String, Double>>() { @Override public Tuple2<String, Double> map(SensorReading value) throws Exception { return new Tuple2<>(value.getId(), value.getTemperature()); } }) .keyBy(data -> data.f0) .timeWindow( Time.seconds(15) ) .minBy(1); 时间间隔可以通过 Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。 2.1.2、滑动窗口（SlidingEventTimeWindows） 滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是 window_size，一个是 sliding_size。 下面代码中的 sliding_size 设置为了 5s，也就是说，每 5s 就计算输出结果一次，每一次计算…

Flink 流处理API
type: item
date: 2022-03-15T13:10:00+00:00
description: 一、Environment 1.1、getExecutionEnvironment 创建一个执行环境，表示当前执行程序的上下文。 如果程序是独立调用的，则 此方法返回本地执行环境；如果从命令行客户端调用程序以提交到集群，则此方法 返回此集群的执行环境，也就是说，getExecutionEnvironment 会根据查询运行的方 式决定返回什么样的运行环境，是最常用的一种创建执行环境的方式。 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 如果没有设置并行度，会以 flink-conf.yaml 中的配置为准，默认是 1。 parallelism.default = 1 1.2、createLocalEnvironment 返回本地执行环境，需要在调用时指定默认的并行度。 LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1); 1.3、createRemoteEnvironment 返回集群执行环境，将 Jar 提交到远程服务器。需要在调用时指定 JobManager 的 IP 和端口号，并指定要在集群中运行的 Jar 包。 StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname", 6123, "YOURPATH//WordCount.jar"); 二、Source 2.1、从集合读取数据 public class SourceTest1_Collection { public static void main(String[] args) throws Exception{ // 创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 从集合中读取数据 DataStream<SensorReading> dataStream = env.fromCollection(Arrays.asList( new SensorReading("sensor_1", 1547718199L, 35.8), new SensorReading("sensor_6", 1547718201L, 15.4), new SensorReading("sensor_7", 1547718202L, 6.7), new SensorReading("sensor_10", 1547718205L, 38.1) )); DataStream<Integer> integerDataStream = env.fromElements(1, 2, 4, 67, 189); // 打印输出 dataStream.print("data"); integerDataStream.print("int"); // 执行 env.execute(); } } 2.2、从文件读取数据 DataStream<String> dataStream = env.readTextFile("YOUR_FILE_PATH "); 2.3、以 kafka 消息队列的数据作为来源 需要引入 kafka 连接器的依赖： <! -- https://mvnrepository.com/artifa ct/org.apache.flink/flink - connector - kafka - 0.11 -- > <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka-0.11_2.12</artifactId> <version>1.10.1</version> </dependency> 具体代码如下： // kafka 配置项 Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.set…

Flink 部署&运行架构
type: item
date: 2022-03-15T13:04:00+00:00
description: 三、Flink部署 3.1、Standalone 模式 3.1.1、安装 解压缩 flink-1.10.1-bin-scala_2.12.tgz，进入 conf 目录中。 tar -zxvf /opt/software/flink-1.10.1-bin-scala_2.12.tgz -C /opt/module/ 1、修改flink/conf/flink-conf.yaml文件 # The external address of the host on which the JobManager runs and can be # reached by the TaskManagers and any clients which want to connect. This setting # is only used in Standalone mode and may be overwritten on the JobManager side # by specifying the --host <hostname> parameter of the bin/jobmanager.sh executable. # In high availability mode, if you use the bin/start-cluster.sh script and setup # the conf/masters file, this will be taken care of automatically. Yarn/Mesos # automatically configure the host name based on the hostname of the node where the # JobManager runs. jobmanager.rpc.address: hadoop102 2、修改/conf/slaves文件： hadoop103 hadoop104 3、分发到另外两台机器上 xsync flink-1.10.0 4、启动 bin/start-cluster.sh(如果是单机部署，则无需上述的配置) 访问 http://localhost:8081 可以对 flink 集群和任务进行监控管理。 3.1.2、提交任务 1）准备数据文件 （如果需要） 2）把含数据文件的文件夹，分发到 taskmanage 机器中 如果从文件中读取数据，由于是从本地磁盘读取，实际任务会被分发到taskmanage 的机器中，所以要把目标文件分发。 3）执行任务 ./flink run -c com.atguigu.wc.StreamWordCount -p 2 FlinkTutorial-1.0-SNAPSHOT.jar --host localhost -port 7777 在 linux 系统中用 netcat 命令进行发送测试： nc -lk 7777 4）查看计算结果 注意：如果输出到控制台，应该在 taskmanager 下查看；如果计算结果输出到文件，同样会保存到 taskmanage 的机器下，不会在 jobmanage 下。 5）在 webui 控制台查看计算过程 3.2、Yarn 模式 以 Yarn 模式部署 Flink 任务时，要求Flink 是有 Hadoop 支持的版本，Hadoop环境需要保证版本在 2.2 以上，并且集群中安装有 HDFS 服务。 3.2.1、Flink on Yarn Flink 提供了两种在 yarn 上运行的模式，分别为 Session-Cluster 和 Per-Job-Cluster 模式。 1、Session-cluster 模式： Session-Cluster 模式需要先启动集群，然后再提交作业，接着会向 yarn 申请一块空间后，资源永远保持不变。如果资源满了，下一个作业就无法提交，只能等到 yarn 中的其中一个作业执行完成后，释放了资源，下个作业才会正常提交。所有作业共享 Dispatcher 和 ResourceManager；共享资源；适合规模小执行时间短的作业。 在 yarn 中初始化一个 flink 集群，开辟指定的资源，以后提交任务都向这里提交。这个 flink 集群会常驻在 yarn 集群中，除非手工停止。 2、Per-Job-Cluster 模式： 一个 Job 会对应一个集群，每提交一个作业会根据自身的情况，都会单独向 yarn 申请资源，直到作业执行完成，一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享 Dispatcher 和 ResourceManager，按需接受资源申请；适合规模大长时间运行的作业。 每次提交都会创建一个新的 …