聊聊Flink的必知必会(一)

聊聊,flink · 浏览次数 : 58

小编点评

| **Key** | **内容** | |---|---| | **Flink** | Flink 是一个分布式计算框架,用于处理大型数据集。 | | **KeyBy()** | 将单词作为Key,把相同单词都发送到同一分区上。 | | **Broadcast** | 将某份数据发送到所有分区上。 | | **Random** | 将所有数据随机均匀地发送到多个分区上,以保证数据平均分配到各个分区。 | | **TaskManager** | 负责执行计算任务,启动时会将Slot资源向ResourceManager注册。 | | **Slot** | 每个TaskManager可以管理一个或多个Slot。 | | **Task** | 每个Slot上可以有多个线程。 |

正文

概述

Flink 是一个框架和分布式处理引擎,用于在无边界和有边界数据流上进行有状态的计算。Flink能在所有常见集群环境中运行,并能以内存速度和任意规模进行计算。使用官网的语句来介绍, Flink 就是 “Stateful Computations over Data Streams”。

首先,Flink是一个纯流式的计算引擎,它的基本数据模型是数据流。流可以是无边界的无限流,即一般意义上的流处理。也可以是有边界的有限流,也就是批处理。因此 Flink 用一套架构同时支持了流处理和批处理。其次,Flink支持有状态的计算,稍微复杂一点的数据处理,比如说基本的聚合,数据流之间的关联都是有状态处理。

更直观的讲,对 Flink ⽽⾔,其所要处理的主要场景就是流数据,批数据只是流数据的⼀个极限特例⽽已,所以 Flink 是⼀款真正的流批统⼀的计算引擎。

无界数据
无界数据是持续产生的数据,所以必须持续地处理无界数据流。数据是无限的,也就无法等待所有输入数据到达后处理,因为输入是无限的,没有终止的时间。处理无界数据通常要求以特定顺序(例如事件发生的顺序)获取,以便判断事件是否完整、有无遗漏。

有界数据
有界数据,就是在一个确定的时间范围内的数据流,有开始有结束,一旦确定了就不会再改变。

Flink流程执行视图

Flink示例Demo可参考
Flink对接KafKa消费分词统计Demo

数据流视图

Flink的程序一般可以分为三个部分:

///// Source
//将kafka和zookeeper配置信息加载到Flink的执行环境当中StreamExecutionEnvironment
FlinkKafkaConsumer08<String> myConsumer = new FlinkKafkaConsumer08<String>("test", new SimpleStringSchema(),
        properties);

///// Transformation
//添加数据源,此处选用数据流的方式,将KafKa中的数据转换成Flink的DataStream类型
DataStream<String> stream = env.addSource(myConsumer);
//流计算,也是Flink的算子部分keyBy是通过key的形式进行分组,此处传入的0,表示是通过二元组的第一个元素也就是单词进行分组
//LineSplitter是算子的具体逻辑
DataStream<Tuple2<String, Integer>> counts = stream.flatMap(new LineSplitter())
        .keyBy(0)
        .sum(1);

///// Sink
//打印输出
counts.print();

第1部分读取数据源(Source);可来自文本、kafka等
第2部分对数据做转换操作(Transformation);其实就是函数式编程中一系列的API;大差不差,毕竟Flink本身就是基于函数式语言Scala编写面向函数式操作的。
第3部分将转换结果输出到一个目的地(Sink);文本、其它存储。

在Transformation部分会有一系列相关的函数算子(Operator)参与对数据的计算,整个链路的逻辑视图可按如下表示:

image

既然有逻辑视图那就会有物理视图,物理视图对应的是分布式环境中具体的执行链路,可参考如下:

image

在分布式计算环境下,执行计算的单个节点(物理机或虚拟机)被称为实例,一个算子在并行执行时,算子子任务会分布到多个节点上,所以算子子任务又被称为算子实例(Instance)。即使输入数据增多,我们也可以通过部署更多的算子子任务来进行横向扩展。除去Sink外的算子都被分成了2个算子子任务,它们的并行度(Parallelism)为2,Sink算子的并行度为1。并行度是可以被设置的,当设置某个算子的并行度为2时,也就意味着这个算子有2个算子子任务(或者说2个算子实例)并行执行。实际应用中一般根据输入数据量的大小、计算资源的多少等多方面的因素来设置并行度。

算子的并行度设置:
env.setParallelism(2);
Sink算子的并行度设置成了1
wordCount.print().setParallelism(1);

数据交换策略

图中出现了数据流动的现象,即数据在不同的算子子任务上进行数据交换。无论是Hadoop、Spark还是Flink,都会涉及数据交换策略。常见的数据交换策略有4种,如下所示。

image

  • 前向传播(Forward)
    前一个算子子任务将数据直接传递给后一个算子子任务,数据不存在跨分区的交换,也避免了因数据交换产生的各类开销,图中Source和FlatMap之间就是这样的情形。

  • 按Key分组(Key-Based)
    数据以(Key,Value)二元组形式存在,该策略将所有数据按照Key进行分组,相同Key的数据会被分到一组、发送到同一个分区上。WordCount程序中,keyBy()将单词作为Key,把相同单词都发送到同一分区,以方便后续算子的聚合统计。

  • 广播(Broadcast)
    将某份数据发送到所有分区上,这种策略涉及了数据在全局的复制,因此非常消耗资源。

  • 随机(Random)
    该策略将所有数据随机均匀地发送到多个分区上,以保证数据平均分配到不同分区上。该策略通常为了防止数据倾斜到某些分区,导致部分分区数据稀疏,另外一些分区数据拥堵的情况发生。

分布式架构与核心组件

为了支持分布式执行,Flink跟其他大数据框架一样,采用了主从(Master-Worker)架构。Flink执行时主要包括如下两个组件:

  1. Master是一个Flink作业的主进程。它起到了协调管理的作用。

  2. TaskManager,又被称为Worker或Slave,是执行计算任务的进程。它拥有CPU、内存等计算资源。Flink作业需要将计算任务分发到多个TaskManager上并行执行。

Flink支持多种部署模式,可以部署在单机(Local)、集群(Cluster),以及云(Cloud)上。

  • Local模式:一般是单节点;
  • Cluster模式:一般使用Cluster模式将Flink作业投入到生产环境中,生产环境可以是Standalone的独立集群,也可以是YARN或Kubernetes集群。
  • Cloud模式:Flink也可以部署在各大云平台上,包括AWS、谷歌云和阿里云。

Standalone集群上只运行Flink作业。除了Flink,绝大多数企业的生产环境运行包括MapReduce、Spark等各种各样的计算任务,一般都会使用YARN或Kubernetes等方式对计算资源进行管理和调度。Flink目前已经支持了YARN、Mesos以及Kubernetes,开发者提交作业的方式变得越来越简单。

Flink作业提交

请先了解集群模式提交作业的流程Flink学习(一):SocketWindowWordCount示例

以Standalone集群部署为例,分析作业的提交过程。

image

如下所述:
①用户编写应用程序代码,并通过Flink客户端(Client)提交作业。程序一般为Java或Scala语言,调用Flink API,构建逻辑视图。代码和相关配置文件被编译打包,被提交到Master的Dispatcher,形成一个应用作业(Application)。

②Dispatcher接收到这个作业,启动JobManager,这个JobManager会负责本次作业的各项协调工作。

③JobManager向ResourceManager申请本次作业所需资源。

④由于在一开始启动Master和TaskManager等进程,TaskManager已经向ResourceManager中注册了资源,这时闲置的TaskManager会被反馈给JobManager。

⑤JobManager将用户作业中的逻辑视图转化为如上所示的并行化的物理执行图,将计算任务分发部署到多个TaskManager上。至此,一个Flink作业就开始执行了。

TaskManager在执行计算任务过程中可能会与其他TaskManager交换数据,会使用一些数据交换策略。同时,TaskManager也会将一些任务状态信息反馈给JobManager,这些信息包括任务启动、执行或终止的状态,快照的元数据等。

Flink核心组件

Client

用户一般使用Client提交作业,比如Flink主目录下bin目录中提供的命令行工具。Client会对用户提交的Flink作业进行预处理,并把作业提交到Flink集群上。Client提交作业时需要配置一些必要的参数,比如使用Standalone集群还是YARN集群等。整个作业被打成了JAR包,DataStream API被转换成了JobGraph,JobGraph是一种类似如上所示的逻辑视图。

Dispatcher

Dispatcher可以接收多个作业,每接收一个作业,Dispatcher都会为这个作业分配一个JobManager。Dispatcher对外提供一个表述性状态转移(Representational State Transfer,REST)式的接口,以超文本传输协议(Hyper Text Transfer Protocal,HTTP)来对外提供服务。

JobManager

JobManager是单个Flink作业的协调者,一个作业会有一个JobManager来负责。JobManager会将Client提交的JobGraph转化为ExecutionGraph,ExecutionGraph是类似如上所示的并行的物理执行图。JobManager会向ResourceManager申请必要的资源,当获取足够的资源后,JobManager将ExecutionGraph以及具体的计算任务分发部署到多个TaskManager上。同时,JobManager还负责管理多个TaskManager,包括收集作业的状态信息、生成检查点、必要时进行故障恢复等。

ResourceManager

Flink现在可以部署在Standalone、YARN或Kubernetes等环境上,不同环境中对计算资源的管理模式略有不同,Flink使用一个名为ResourceManager的模块来统一处理资源分配上的问题。在Flink中,计算资源的基本单位是TaskManager上的任务槽位(Task Slot,简称Slot)。ResourceManager的职责主要是从YARN等资源提供方获取计算资源,当JobManager有计算需求时,将空闲的Slot分配给JobManager。当计算任务结束时,ResourceManager还会重新收回这些Slot。

TaskManager

TaskManager是实际负责执行计算的节点。一般地,一个Flink作业是分布在多个TaskManager上执行的,单个TaskManager上提供一定量的Slot。一个TaskManager启动后,相关Slot信息会被注册到ResourceManager中。当某个Flink作业提交后,ResourceManager会将空闲的Slot提供给JobManager。JobManager获取到空闲的Slot后会将具体的计算任务部署到空闲Slot之上,任务开始在这些Slot上执行。在执行过程,由于要进行数据交换,TaskManager还要和其他TaskManager进行必要的数据通信。总之,TaskManager负责具体计算任务的执行,启动时它会将Slot资源向ResourceManager注册。

任务执行与资源划分

任务执行

Flink采用主从架构,Master起着管理协调作用,TaskManager负责物理执行,在执行过程中会发生一些如数据交换、生命周期管理等事情。用户调用Flink API,构造逻辑视图,Flink会对逻辑视图优化,并转化为并行化的物理执行图,最后被执行的是物理执行图。

在构造物理执行图的过程中,Flink会将一些算子子任务链接在一起,组成算子链。链接后以任务(Task)的形式被TaskManager调度执行。使用算子链是一个非常有效的优化,它可以有效减少算子子任务之间的传输开销。链接之后形成的任务是TaskManager中的一个线程。

image

例如,数据从Source前向传播到FlatMap,这中间没有发生跨分区的数据交换,因此,我们完全可以将Source、FlatMap这两个子任务组合在一起,形成一个任务。数据经过keyBy()发生了数据交换,数据会跨越分区,因此无法将keyBy()以及其后面的窗口聚合、链接到一起。由于WindowAggregation的并行度为2、Sink的并行度为1,数据再次发生了交换,我们不能把WindowAggregation和Sink两部分链接到一起。

默认情况下,Flink会尽量将更多的子任务链接在一起,这样能减少一些不必要的数据传输开销。但一个子任务有超过一个输入或发生数据交换时,链接就无法建立。

Slot

TaskManager是一个JVM进程,在TaskManager中可以并行执行一到多个任务。每个任务是一个线程,需要TaskManager为其分配相应的资源,TaskManager使用Slot给任务分配资源。

一个TaskManager是一个进程,TaskManager可以管理一至多个任务,每个任务是一个线程,占用一个Slot。每个Slot的资源是整个TaskManager资源的子集,下图的TaskManager下有3个Slot,每个Slot占用TaskManager 1/3的内存,第一个Slot中的任务不会与第二个Slot中的任务互相争抢内存资源;即线程隔离。

image

Flink允许用户设置TaskManager中Slot的数目,这样用户就可以确定以怎样的粒度将任务做相互隔离。如果每个TaskManager只包含一个Slot,那么该Slot内的任务将独享JVM。如果TaskManager包含多个Slot,那么多个Slot内的任务可以共享JVM资源,比如共享TCP连接、心跳信息、部分数据结构等。官方建议将Slot数目设置为TaskManager下可用的CPU核心数,那么平均下来,每个Slot都能获得1个CPU核心。

针对Slot槽位的分配有多种方式,比如可以使用槽位共享等,进一步减少数据传输开销,充分利用计算资源。

image

开启槽位共享后,Flink允许多个任务共享一个Slot。如图所示,最左侧的数据流,一个作业从Source到Sink的所有子任务都可以放置在一个Slot中,这样数据交换成本更低。而且,对于一个数据流图来说,Source、FlatMap等算子的计算量相对不大,WindowAggregation算子的计算量比较大,计算量较大的算子子任务与计算量较小的算子子任务可以互补,空出更多的槽位,分配给更多任务,这样可以更好地利用资源。

总结

简单的介绍了从Flink的相关概念及关键架构及组件,对此有个基本的印象概念;强烈建议要先走一遍官方的demo,尤其是集群模式提交作业,才能有更立体更深入的了解。

Flink的官方SocketWindowWordCount单词计数demo体验

Flink 大数据初体验(hello world案例)

与聊聊Flink的必知必会(一)相似的内容:

聊聊Flink的必知必会(一)

Flink 是一个框架和分布式处理引擎,用于在无边界和有边界数据流上进行有状态的计算。Flink能在所有常见集群环境中运行,并能以内存速度和任意规模进行计算。

聊聊Flink的必知必会(三)

### 概述 在进行流处理时,很多时候想要对流的有界子集进行聚合分析。例如有如下的需求场景: (1)每分钟的页面浏览(PV)次数。 (2)每用户每周的会话次数。 (3)每分钟每传感器的最高温度。 (4)当电商发布一个秒杀活动时,想要每隔10min了解流量数据。 对于这些需求的处理,程序需要处理元素组

聊聊Flink CDC必知必会

CDC是(Change Data Capture变更数据获取)的简称。 核心思想是,监测并捕获数据库的变动(包括数据 或 数据表的插入INSERT、更新UPDATE、删除DELETE等),将这些变更按发生的顺序完整记录下来,写入到消息中间件中以供其他服务进行订阅及消费。 ## Flink CDC的设

聊聊Flink必知必会(二)

### Checkpoint与Barrier Flink是一个有状态的流处理框架,因此需要对状态做持久化,Flink定期保存状态数据到存储空间上,故障发生后从之前的备份中恢复,这个过程被称为Checkpoint机制。而Checkpoint为Flink提供了Exactly-Once的投递保障。 流处理

聊聊Flink必知必会(四)

### 概述 Flink Streaming API借鉴了谷歌数据流模型(Google Data Flow Model),它的流API支持不同的时间概念。Flink明确支持以下3个不同的时间概念。 Flink明确支持以下3个不同的时间概念。 (1)事件时间:事件发生的时间,由产生(或存储)事件的设备

聊聊日志聚类算法及其应用场景

阅读《[基于 Flink ML 搭建的智能运维算法服务及应用](https://mp.weixin.qq.com/s/yhXiQtUSR4hxp9XWrkiiew "基于 Flink ML 搭建的智能运维算法服务及应用")》一文后,对其中日志聚类算法有了些思考。 ### 概述 日志聚类,简而言之是对

聊聊GLM-4-9B开源模型的微调loss计算

概述 Github官方地址:GLM-4 网上已经有很多关于微调的文章,介绍各种方式下的使用,这里不会赘述。我个人比较关心的是微调时的loss计算逻辑,这点在很多的文章都不会有相关的描述,因为大多数人都是关心如何使用之类的应用层,而不是其具体的底层逻辑,当然咱也说不清太底层的计算。 可了解其它loss

聊聊一个差点被放弃的项目以及近期的开源计划

前言 自从 StarBlog 和 SiteDirectory 之后,我还没写新的关于开源项目的系列,最近又积累了很多想法,正好写一篇博客来总结一下。 关于差点被放弃的项目,就是最近一直在做的单点认证(IdentityServerLite) IdentityServerLite 开发这个项目的起因,是

聊聊 JSON Web Token (JWT) 和 jwcrypto 的使用

哈喽大家好,我是咸鱼。 最近写的一个 Python 项目用到了 jwcrypto 这个库,这个库是专门用来处理 JWT 的,JWT 全称是 JSON Web Token ,JSON 格式的 Token。 今天就来简单入门一下 JWT。 官方介绍:https://jwt.io/introduction

聊聊MySQL是如何处理排序的

在MySQL的查询中常常会用到 order by 和 group by 这两个关键字,它们的相同点是都会对字段进行排序,那查询语句中的排序是如何实现的呢?