大数据Hadoop之——实时计算流计算引擎Flink（Flink环境部署）

文章目录

一、概述二、Flink工作原理三、Flink核心概念

1）Time（时间语义）2）Window（窗口）3） Trigger

1、自定义触发器2、预定义触发器 4）State5）状态存储

1、MemoryStateBackend2、FsStateBackend3、RocksDBStateBackend 6）Checkpoint

1、Asynchronous Barrier Snapshots(ABS)2、Exactly-once vs At-Least-Once 7）Watermark

1、自定义数据源设置 Timestamp/Watermark2、在数据流中设置 Timestamp/Watermark 8）广播状态（Broadcast State）9）Operator Chain10）Side Output 四、对比常用的实时计算框架五、Flink环境部署（Flink On Yarn）

1）Local模式2）Standalone模式

1、机器及角色划分1、下载2、修改配置文件3、将安装目录复制到另外两台节点4、配置环境变量，修改/etc/profile5、启动集群 3）On Yarn模式（推荐）

1、yarn-session模式2、yarn-cluster模式（推荐）

historyserver简介与配置六、Spark与Flink对比

一、概述

Flink核心是一个流式的数据流执行引擎，并且能够基于同一个Flink运行时，提供支持流处理和批处理两种类型应用。其针对数据流的分布式计算提供了数据分布，数据通信及容错机制等功能。基于流执行引擎，Flink提供了跟多高抽象层的API便于用户编写分布式任务，下面稍微介绍一下Flink的几种API：

Flink官网：https://flink.apache.org/
官方文档（1.14.2版本）：https://nightlies.apache.org/flink/flink-docs-release-1.14/
官方中文文档（1.14.2版本）：https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/
不同版本的文档：https://nightlies.apache.org/flink/

SQL & Table API：Flink 支持两种关系型的 API，Table API 和 SQL。这两个 API 都是批处理和流处理统一的 API，这意味着在无边界的实时数据流和有边界的历史记录数据流上，关系型 API 会以相同的语义执行查询，并产生相同的结果。Table API 和 SQL借助了 Apache Calcite 来进行查询的解析，校验以及优化。它们可以与 DataStream 和DataSet API 无缝集成，并支持用户自定义的标量函数，聚合函数以及表值函数。DataStream API：DataStream API为许多通用的流处理操作提供了处理原语。这些操作包括窗口、逐条记录的转换操作，在处理事件时进行外部数据库查询等。DataStream API 支持 Java 和Scala 语言，预先定义了例如 map()、reduce()、aggregate() 等函数。你可以通过扩展实现预定义接口或使用 Java、Scala 的 lambda 表达式实现自定义的函数。DataSet API：DataSet API 是 Flink 用于批处理应用程序的核心 API。DataSet API 所提供的基础算子包括 map、reduce、(outer) join、co-group、iterate 等。所有算子都有相应的算法和数据结构支持，对内存中的序列化数据进行操作。如果数据大小超过预留内存，则过量数据将存储到磁盘。Flink 的 DataSet API 的数据处理算法借鉴了传统数据库算法的实现，例如混合散列连接（hybrid hash-join）和外部归并排序（external merge-sort）。StateFul Stream Processing：最低级抽象只提供有状态流，通过Process Function嵌入到DataStream API中，它允许用户自由处理来自一个或者多个流的时间，并使用一致的容错状态，此外用户可以注册event time和processing time回调，允许程序实现复杂的计算。扩展库

复杂事件处理(CEP)：模式检测是事件流处理中的一个非常常见的用例。Flink 的 CEP库提供了 API，使用户能够以例如正则表达式或状态机的方式指定事件模式。CEP 库与Flink 的 DataStream API 集成，以便在 DataStream 上评估模式。CEP 库的应用包括网络入侵检测，业务流程监控和欺诈检测。Gelly: Gelly 是一个可扩展的图形处理和分析库。Gelly 是在 DataSet API 之上实现的，并与 DataSet API 集成。因此，它能够受益于其可扩展且健壮的操作符。Gelly 提供了内置算法，如 label propagation、triangle enumeration 和 page rank 算法，也提供了一个简化自定义图算法实现的 Graph API。二、Flink工作原理

Flink的基础架构图：

JobClient：负责接收程序，解析和优化程序的执行计划，然后提交执行计划到JobManager。这里执行的程序优化是将相邻的Operator融合，形成Operator Chain，Operator的融合可以减少task的数量，提高TaskManager的资源利用率。为了了解Flink的解析过程，需要简单介绍一下Flink的Operator，在Flink主要有三类Operator：

Source Operator ：顾名思义这类操作一般是数据来源操作，比如文件、socket、kafka等，一般存在于程序的最开始Transformation Operator：这类操作主要负责数据转换，map，flatMap，reduce等算子都属于Transformation Operator，Sink Operator：意思是下沉操作，这类操作一般是数据落地，数据存储的过程，放在Job最后，比如数据落地到Hdfs、Mysql、Kafka等等。

JobManagers：负责申请资源，协调以及控制整个job的执行过程，具体包括，调度任务、处理checkpoint、容错等等。

TaskManager：TaskManager运行在不同节点上的JVM进程，负责接收并执行JobManager发送的task,并且与JobManager通信，反馈任务状态信息，如果说JobManager是master的话，那么TaskManager就是worker用于执行任务。每个TaskManager像是一个容器，包含一个或者多个Slot。

Slot：Slot是TaskManager资源粒度的划分，每个Slot都有自己独立的内存。所有Slot平均分配TaskManager的内存，值得注意的是，Slot仅划分内存，不涉及CPU的划分，即CPU是共享使用。每个Slot可以运行多个task。Slot的个数就代表了一个程序的最高并行度。

Task：Task是在operators的subtask进行链化之后形成的，具体Flink job中有多少task和operator的并行度和链化的策略有关。

SubTask：因为Flink是分布式部署的，程序中的每个算子，在实际执行中被分隔为一个或者多个subtask，运算符子任务(subtask)的数量是该特定运算符的并行度。数据流在算子之间流动，就对应到SubTask之间的数据传输。Flink允许同一个job中来自不同task的subtask可以共享同一个slot。每个slot可以执行一个并行的pipeline。可以将pipeline看作是多个subtask的组成的。

三、Flink核心概念 1）Time（时间语义）

Flink 中的 Time 分为三种：事件时间、达到时间与处理时间。

事件时间：是事件真实发生的时间。

达到时间：是系统接收到事件的时间，即服务端接收到事件的时间。

处理时间：是系统开始处理到达事件的时间。

【温馨提示】在某些场景下，处理时间等于达到时间。因为处理时间没有乱序的问题，所以服务端做基于处理时间的计算是比较简单的，无迟到与乱序数据。

Flink 中只需要通过 env 环境变量即可设置Time：

//创建环境上下文
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 设置在当前程序中使用 ProcessingTime
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

2）Window（窗口）

窗口本质就是将无限数据集沿着时间（或者数量）的边界切分成有限数据集。

Time Window：基于时间的，分为Tumbling Window（无数据重叠）和Sliding Window（有数据重叠）。

Count Window：基于数量的，分为Tumbling Window（无数据重叠）和Sliding Window（有数据重叠）。

Session Window：基于会话的，一个session window关闭通常是由于一段时间没有收到元素。

Global Window：全局窗口。

【温馨提示】在实际操作中，window又分为两大类型的窗口：Keyed Window 和 Non-keyed Window，两种类型的窗口操作API有细微的差别。

3） Trigger 1、自定义触发器

触发器决定了窗口何时会被触发计算，Flink 中开发人员需要在 window 类型的操作之后才能调用 trigger 方法传入触发器定义。Flink 中的触发器定义需要继承并实现 Trigger 接口，该接口有以下方法：

onElement()：每个被添加到窗口中的元素都会被调用onEventTime()：当事件时间定时器触发时会被调用，比如watermark到达onProcessingTime()：当处理时间定时器触发时会被调用，比如时间周期触发onMerge()：当两个窗口合并时两个窗口的触发器状态将会被调动并合并clear()：执行需要清除相关窗口的事件

以上方法会返回决定如何触发执行的 TriggerResult：

CONTINUE：什么都不做FIRE：触发计算PURGE：清除窗口中的数据FIRE_AND_PURGE：触发计算后清除窗口中的数据 2、预定义触发器

如果开发人员未指定触发器，则 Flink 会自动根据场景使用默认的预定义好的触发器。在基于事件时间的窗口中使用 EventTimeTrigger，该触发器会在watermark通过窗口边界后立即触发（即watermark出现关闭改窗口时）。在全局窗口（GlobalWindow）中使用 NeverTrigger，该触发器永远不会触发，所以在使用全局窗口时用户需要自定义触发器。

4）State

Managed State 是由flink runtime管理来管理的，自动存储、自动恢复，在内存管理上有优化机制。且Managed State 支持常见的多种数据结构，如value、list、map等，在大多数业务场景中都有适用之处。总体来说是对开发人员来说是比较友好的，因此 Managed State 是 Flink 中最常用的状态。Managed State 又分为 Keyed State 和 Operator State 两种。Raw State 由用户自己管理，需要序列化，只能使用字节数组的数据结构。Raw State 的使用和维度都比 Managed State 要复杂，建议在自定义的Operator场景中酌情使用。 5）状态存储

Flink中状态的实现有三种：MemoryState、FsState、RocksDBState。三种状态存储方式与使用场景各不相同，详细介绍如下：

1、MemoryStateBackend

构造函数：MemoryStateBackend(int maxStateSize, boolean asyncSnapshot)

存储方式：State存储于各个 TaskManager内存中，Checkpoint存储于 JobManager内存

容量限制：单个State最大5M、maxStateSize<=akka.framesize（10M）、总大小不超过JobManager内存

使用场景：无状态或者JobManager挂掉不影响的测试环境等，不建议在生产环境使用

2、FsStateBackend

构造函数：FsStateBackend(URI checkpointUri, boolean asyncSnapshot)

存储方式：State存储于 TaskManager内存，Checkpoint存储于外部文件系统（本次磁盘 or HDFS）

容量限制：State总量不超过TaskManager内存、Checkpoint总大小不超过外部存储空间

使用场景：常规使用状态的作业，分钟级的窗口聚合等，可在生产环境使用

3、RocksDBStateBackend

构造函数：RocksDBStateBackend(URI checkpointUri, boolean enableincrementCheckpoint)

存储方式：State存储于 TaskManager上的kv数据库（内存+磁盘），Checkpoint存储于外部文件系统（本次磁盘 or HDFS）

容量限制：State总量不超过TaskManager内存+磁盘、单key最大2g、Checkpoint总大小不超过外部存储空间

使用场景：超大状态的作业，天级的窗口聚合等，对读写性能要求不高的场景，可在生产环境使用

根据业务场景需要用户选择最合适的 StateBackend ，代码中只需在相应的 env 环境中设置即可：

// flink 上下文环境变量
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 设置状态后端为 FsStateBackend，数据存储到 hdfs /tmp/flink/checkpoint/test 中
env.setStateBackend(new FsStateBackend("hdfs://ns1/tmp/flink/checkpoint/test", false))

6）Checkpoint

Checkpoint 是分布式全域一致的，数据会被写入hdfs等共享存储中。且其产生是异步的，在不中断、不影响运算的前提下产生。

用户只需在相应的 env 环境中设置即可：

// 1000毫秒进行一次 Checkpoint 操作
env.enableCheckpointing(1000)
// 模式为准确一次
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
// 两次 Checkpoint 之间最少间隔 500毫秒
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
// Checkpoint 过程超时时间为 60000毫秒，即1分钟视为超时失败
env.getCheckpointConfig.setCheckpointTimeout(60000)
// 同一时间只允许1个Checkpoint的操作在执行
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

1、Asynchronous Barrier Snapshots(ABS)

异步屏障快照算法，这个算法基本上是Chandy-Lamport算法的变体，针对DAG（有向无环图）的ABS算法执行流程如下所示：

Barrier周期性的被注入到所有的Source中，Source节点看到Barrier后，会立即记录自己的状态，然后将Barrier发送到Transformation Operator。当Transformation Operator从某个input channel收到Barrier后，它会立刻Block住这条通道，直到所有的input channel都收到Barrier，这个等待的过程就叫做屏障对齐（barrier alignment），此时该Operator就会记录自身状态，并向自己的所有output channel广播Barrier。Sink接受Barrier的操作流程与Transformation Oper一样。当所有的Barrier都到达Sink之后，并且所有的Sink也完成了Checkpoint，这一轮Snapshot就完成了。

下面这个图展示了一个ABS算法的执行过程：

2、Exactly-once vs At-Least-once

上面讲到的屏障对齐过程是Flink exactly-once语义的基础，因为屏障对齐能够保证多输入流的算子正常处理不同checkpoint区间的数据，避免它们发生交叉，即不会有数据被处理两次。但是对齐过程需要时间，有一些对延迟特别敏感的应用可能对准确性的要求没有那么高。所以Flink也允许在StreamExecutionEnvironment.enableCheckpointing()方法里指定At-Least-Once语义，会取消屏障对齐，即算子收到第一个输入的屏障之后不会阻塞，而是触发快照。这样一来，部分属于检查点n + 1的数据也会包括进检查点n的数据里，当恢复时，这部分交叉的数据就会被重复处理。 7）Watermark

Flink 程序并不能自动提取数据源中哪个字段/标识为数据的事件时间，从而也就无法自己定义 Watermark 。开发人员需要通过 Flink 提供的 API 来提取和定义 Timestamp/Watermark，可以在数据源或者数据流中定义。

1、自定义数据源设置 Timestamp/Watermark

自定义的数据源类需要继承并实现 SourceFunction[T] 接口，其中 run 方法是定义数据生产的地方：

//自定义的数据源为自定义类型MyType
class MySource extends SourceFunction[MyType]{
    //重写run方法，定义数据生产的逻辑
    override def run(ctx: SourceContext[MyType]): Unit = {
        while () {
            val next: MyType = getNext()
            //设置timestamp从MyType的哪个字段获取(eventTimestamp)
            ctx.collectWithTimestamp(next, next.eventTimestamp)

            if (next.hasWatermarkTime) {
                //设置watermark从MyType的那个方法获取(getWatermarkTime)
                ctx.emitWatermark(new Watermark(next.getWatermarkTime))
            }
        }
    }
}

2、在数据流中设置 Timestamp/Watermark

在这里插入代码片在数据流中，可以设置 stream 的 Timestamp Assigner ，该 Assigner 将会接收一个 stream，并生产一个带 Timestamp和Watermark 的新 stream。

val withTimestampsAndWatermarks: DataStream[MyEvent] = stream.assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks())

8）广播状态（Broadcast State）

和 Spark 中的广播变量一样，Flink 也支持在各个节点中各存一份小数据集，所在的计算节点实例可在本地内存中直接读取被广播的数据，可以避免Shuffle提高并行效率。

广播状态（Broadcast State）的引入是为了支持一些来自一个流的数据需要广播到所有下游任务的情况，它存储在本地，用于处理其他流上的所有传入元素。

// key the shapes by color
KeyedStream colorPartitionedStream = shapeStream.keyBy(new KeySelector(){...});

// a map descriptor to store the name of the rule (string) and the rule itself.
MapStateDescriptor ruleStateDescriptor = new MapStateDescriptor<>("RulesBroadcastState",BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(new TypeHint() {}));
        
// broadcast the rules and create the broadcast state
BroadcastStream ruleBroadcastStream = ruleStream.broadcast(ruleStateDescriptor);

DataStream output = colorPartitionedStream.connect(ruleBroadcastStream).process(new KeyedBroadcastProcessFunction(){...});

9）Operator Chain

Flink作业中，可以指定相关的chain将相关性非常强的转换操作（operator）绑定在一起，使得上下游的Task在同一个Pipeline中执行，避免因为数据在网络或者线程之间传输导致的开销。

一般情况下Flink在Map类型的操作中默认开启 Operator Chain 以提高整体性能，开发人员也可以根据需要创建或者禁止 Operator Chain 对任务进行细粒度的链条控制。

//创建 chain
dataStream.filter(...).map(...).startNewChain().map(...)
//禁止 chain
dataStream.map(...).disableChaining()

创建的链条只对当前的操作符和之后的操作符有效，不不影响其他操作，如上代码只针对两个map操作进行链条绑定，对前面的filter操作无效，如果需要可以在filter和map之间使用 startNewChain方法即可。

10）Side Output

除了从DataStream操作的结果中获取主数据流之外，Flink还可以产生任意数量额外的侧输出(Side Output)结果流。侧输出结果流的数据类型不需要与主数据流的类型一致，不同侧输出流的类型也可以不同。当要拆分数据流时(通常必须复制流)，从每个流过滤出不想拥有的数据时Side Output将非常有用。

DataStream input = ...;
final OutputTag outputTag = new OutputTag("side-output"){};
SingleOutputStreamOperator mainDataStream = input
  .process(new ProcessFunction() {
      @Override
      public void processElement(
          Integer value,
          Context ctx,
          Collector out) throws Exception {
        // 将数据发送到常规输出中
        out.collect(value);
        // 将数据发送到侧输出中
        ctx.output(outputTag, "sideout-" + String.valueOf(value));
      }
    });
DataStream sideOutputStream = mainDataStream.getSideOutput(outputTag);

四、对比常用的实时计算框架

Flink 是有状态的和容错的，可以在维护一次应用程序状态的同时无缝地从故障中恢复。它支持大规模计算能力，能够在数千个节点上并发运行。它具有很好的吞吐量和延迟特性。同时，Flink 提供了多种灵活的窗口函数。Flink 在流式计算里属于真正意义上的单条处理，每一条数据都触发计算，而不是像 Spark 一样的 Mini Batch 作为流式处理的妥协。Flink的容错机制较为轻量，对吞吐量影响较小，而且拥有图和调度上的一些优化，使得 Flink 可以达到很高的吞吐量。而 Strom 的容错机制需要对每条数据进行ack，因此其吞吐量瓶颈也是备受诟病。五、Flink环境部署（Flink On Yarn）

下载地址：http://flink.apache.org/downloads.html

1）Local模式

对于 Local 模式来说，JobManager 和 TaskManager 会公用一个 JVM 来完成 Workload。如果要验证一个简单的应用，Local 模式是最方便的。实际企业中大多使用Flink On Yarn模式，而local模式只是将安装包解压启动（./bin/start-cluster.sh）即可。其实local模式就是单节点，master和woker节点都是同一台机器。

Local Cluster模式是开箱即用的，直接解压安装包，然后启动即可。

$ cd /opt/bigdata/hadoop/software
$ wget https://dlcdn.apache.org/flink/flink-1.14.2/flink-1.14.2-bin-scala_2.12.tgz
# 解压
$ tar -zxvf flink-1.14.2-bin-scala_2.12.tgz -C /opt/bigdata/hadoop/server/
# 进入bin目录运行启动脚本
$ cd /opt/bigdata/hadoop/server/flink-1.14.2
$ ./bin/start-cluster.sh

打开浏览器输入http://IP:8081，查看WEBUI监控界面
我这里地址：http://hadoop-node1:8081

2）Standalone模式

Stanalone CLuster是一种独立的集群模式，集群运行不需要依赖外部系统，完全自己独立进行管理。

1、机器及角色划分

机器IP	机器名	节点类型
192.168.0.113	hadoop-node1	Master
192.168.0.114	hadoop-node2	Worker
192.168.0.115	hadoop-node3	Worker

1、下载

$ cd /opt/bigdata/hadoop/software
$ wget https://dlcdn.apache.org/flink/flink-1.14.2/flink-1.14.2-bin-scala_2.12.tgz
# 解压
$ tar -zxvf flink-1.14.2-bin-scala_2.12.tgz -C /opt/bigdata/hadoop/server/
$ cd /opt/bigdata/hadoop/server/flink-1.14.2

2、修改配置文件

修改flink-conf.yaml文件

$ cd /opt/bigdata/hadoop/server/flink-1.14.2/conf
$ vi flink-conf.yaml

## jobmanager节点地址，也是master节点地址
jobmanager.rpc.address: hadoop-node1

其它使用默认配置，其中有一些HA高可用、容错、安全、HistoryServer相关配置，按需进行配置即可，HistoryServer需单独运行启动脚本来启动服务。

修改masters文件
把默认的localhost:8081删掉，添加如下内容：

hadoop-node1:8081

修改workers文件，内容如下：
把默认的localhost删掉，添加如下内容：

hadoop-node2
hadoop-node3

3、将安装目录复制到另外两台节点

$ cd /opt/bigdata/hadoop/server
$ scp -r flink-1.14.2 hadoop-node2:/opt/bigdata/hadoop/server/
$ scp -r flink-1.14.2 hadoop-node3:/opt/bigdata/hadoop/server/

4、配置环境变量，修改/etc/profile

在/etc/profile文件中添加如下内容（所有节点）：

export Flink_HOME=/opt/bigdata/hadoop/server/flink-1.14.2
export PATH=$PATH:$Flink_HOME/bin

使配置文件生效

$ source /etc/profile

5、启动集群

$ start-cluster.sh

3）On Yarn模式（推荐）

On Yarn官方文档：https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/deployment/resource-providers/yarn/

YARN模式是使用YARN做为Flink运行平台，JobManager、TaskManager、用户提交的应用程序都运行在YARN上。

Flink on yarn 有两种运行模式：

创建一个yarn-session，后续提交应用至该session直接 yarn cluster模式提交应用至yarn服务

下载

$ cd /opt/bigdata/hadoop/software
$ wget https://dlcdn.apache.org/flink/flink-1.14.2/flink-1.14.2-bin-scala_2.12.tgz
# 解压
$ tar -zxvf flink-1.14.2-bin-scala_2.12.tgz -C /opt/bigdata/hadoop/server/
$ cd /opt/bigdata/hadoop/server/flink-1.14.2

配置
在/etc/profile文件中追加如下内容：

export Flink_HOME=/opt/bigdata/hadoop/server/flink-1.14.2
export PATH=$PATH:$Flink_HOME/bin
# 上面两句如果加过，可以忽略
export HADOOP_CLASSPATH=`hadoop classpath`

加载配置

$ source /etc/profile

1、yarn-session模式

Yarn-session模式下，首先向Yarn提交一个长时运行的空应用，运行起来之后，任务跑完集群也不释放，可以重复使用在Yarn上开启的Flink集群，也称为共享型集群，适合小任务。

实验

第一种模式分为两步：yarn-session.sh(启动，开辟/申请资源)+flink run（提交任务）

【第一步】开源资源，使用如下命令：

$ yarn-session.sh -n 2 -jm 1024 -tm 1024 -d

### 参数解释：
#-n 2 表示指定两个容器 
# -jm 1024 表示jobmanager 1024M内存 
# -tm 1024表示taskmanager 1024M内存 
#-d 任务后台运行 
### 如果你不希望flink yarn client一直运行，也可以启动一个后台运行的yarn session。使用这个参数：-d 或者 --detached。在这种情况下，flink yarn client将会只提交任务到集群然后关闭自己。注意：在这种情况下，无法使用flink停止yarn session，必须使用yarn工具来停止yarn session。
# yarn application -kill $applicationId
#-nm,--name  YARN上为一个自定义的应用设置一个名字
#-q,--query  显示yarn中可用的资源 (内存, cpu核数)
#-z,--zookeeperNamespace    针对HA模式在zookeeper上创建NameSpace
#-id,--applicationId    YARN集群上的任务id，附着到一个后台运行的yarn session中

JobManager Web Interface: http://hadoop-node2:41787，端口是随机的。

通过yarn入口访问flink

【第二步】提交任务
为了进行测试，我们对Flink目录下的LICENSE文件进行词频统计

$ cd $Flink_HOME
$ hadoop fs -put LICENSE /
$ hadoop fs -ls  /LICENSE
# 提交任务
$ flink run ./examples/batch/WordCount.jar -input hdfs://hadoop-node1:8082/LICENSE -output hdfs://hadoop-node1:8082/wordcount-result.txt

再提交一次任务

【注意】-output一定是不存在的文件，有flink自动创建写入

$ flink run ./examples/batch/WordCount.jar -input hdfs://hadoop-node1:8082/LICENSE -output hdfs://hadoop-node1:8082/wordcount-result2.txt

发现现在已经有两个跑完的任务了，但是只有一个flink集群，从而验证了yarn-session模式

2、yarn-cluster模式（推荐）

Yarn-cluster模式下，每个任务单独在Yarn上启动一套Flink集群,适合大任务，运行完后结束，集群释放，资源释放，再有任务，会再起新的Flink集群，需要频繁的在Yanr上开启Flink集群，集群相互独立，适合大任务。

当然除了on yarn模式，还有on k8s，有兴趣的小伙伴，可以试试，当时目前企业里用的最多的还是on yarn模式，但是现在不是流行容器化嘛，以后很大可能会慢慢转到 on k8s模式。

实验

第二种模式其实也分为两个部分，依然是开辟资源和提交任务，但是在Job模式下，这两步都合成一个命令了。

$ cd $Flink_HOME
$ flink run -m yarn-cluster -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar

# 查看帮助
$ flink --help
### 参数详解，这里只列出了部分参数
Options for yarn-cluster mode:
     -d,--detached                        If present, runs the job in detached
                                          mode
     -m,--jobmanager                 Set to yarn-cluster to use YARN
                                          execution mode.
     -yat,--yarnapplicationType      Set a custom application type for the
                                          application on YARN
     -yD                  use value for given property
     -yd,--yarndetached                   If present, runs the job in detached
                                          mode (deprecated; use non-YARN
                                          specific option instead)
     -yh,--yarnhelp                       Help for the Yarn session CLI.
     -yid,--yarnapplicationId        Attach to running YARN session
     -yj,--yarnjar                   Path to Flink jar file
     -yjm,--yarnjobManagerMemory     Memory for JobManager Container with
                                          optional unit (default: MB)
     -ynl,--yarnnodeLabel            Specify YARN node label for the YARN
                                          application
     -ynm,--yarnname                 Set a custom name for the application
                                          on YARN
     -yq,--yarnquery                      Display available YARN resources
                                          (memory, cores)
     -yqu,--yarnqueue                Specify YARN queue.
     -ys,--yarnslots                 Number of slots per TaskManager
     -yt,--yarnship                  Ship files in the specified directory
                                          (t for transfer)
     -ytm,--yarntaskManagerMemory    Memory per TaskManager Container with
                                          optional unit (default: MB)
     -yz,--yarnzookeeperNamespace    Namespace to create the Zookeeper
                                          sub-paths for high availability mode
     -z,--zookeeperNamespace         Namespace to create the Zookeeper
                                          sub-paths for high availability mode

【温馨提示】上面命令中没有指定-input 和 -output，这是由于有默认的数据集和输出方式，看看效果。

发现查看不了History，是因为没起History服务，下面启动这个服务

historyserver简介与配置

History Server允许查询由JobManager归档的已完成作业的状态和统计信息。已完成作业的归档在JobManager上进行，JobManager会将归档的作业信息upload到文件系统目录，这个文件系统可以是本地文件系统、HDFS、H3等，这个目录是可以在配置文件中指定的。然后还需要配置History Server去扫描这个目录，并且可以配置扫描的间隔时间。

配置historyserver

$ cd $Flink_HOME/bin
# 选创建目录
$ hdfs://hadoop-node1:8082/flink/completed-jobs/
# conf/flink-conf.yaml
# 指定由JobManager归档的作业信息所存放的目录，这里使用的是HDFS
jobmanager.archive.fs.dir: hdfs://hadoop-node1:8082/flink/completed-jobs/
# 指定History Server扫描哪些归档目录，多个目录使用逗号分隔
historyserver.archive.fs.dir: hdfs://hadoop-node1:8082/flink/completed-jobs/
# 指定History Server间隔多少毫秒扫描一次归档目录
historyserver.archive.fs.refresh-interval: 10000
# History Server所绑定的ip，hadoop-node1代表允许所有ip访问
historyserver.web.address: hadoop-node1
# 指定History Server所监听的端口号，默认端口是8082
historyserver.web.port: 9082

启动historyserver

$ ./historyserver.sh start
$ jps

web：http://hadoop-node1:9082

重新跑一次任务

$ flink run -m yarn-cluster -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar

查看正在运行任务的日志，yarn容器退出之后，就下面入口就访问不了了。

六、Spark与Flink对比

可以先看我之前写的Spark文章：https://blog.csdn.net/qq_35745940/article/details/122011664

Flink执行流程图如下：

对比维度	Spark	Flink
设计理念	Spark的技术理念是使用微批来模拟流的计算,基于Micro-batch,数据流以时间为单位被切分为一个个批次,通过分布式数据集RDD进行批量处理,是一种伪实时。	Flink是基于事件驱动的，是面向流的处理框架, Flink基于每个事件一行一行地流式处理，是真正的实时流式计算，另外他也可以基于流来模拟批进行计算实现批处理。
架构方面	Spark在运行时的主要角色包括：Master、Worker、Driver、Executor。	Flink 在运行时主要包含：Jobmanager、Taskmanager和Slot。
任务调度	Spark Streaming 连续不断的生成微小的数据批次，构建有向无环图DAG，根据DAG中的action操作形成job，每个job有根据窄宽依赖生成多个stage。	使用DataStream API开发的应用程序，首先被转换为Transformation，再被映射为StreamGraph，在客户端进行StreamGraph、JobGraph的转换，提交JobGraph到Flink集群后，Flink集群负责将JobGraph转换为ExecutionGraph，之后进入调度执行阶段。
时间机制	Spark Streaming 支持的时间机制有限，只支持处理时间。使用processing time模拟event time必然会有误差，如果产生数据堆积的话，误差则更明显。	flink支持三种时间机制：事件时间，注入时间，处理时间，同时支持 watermark 机制处理迟到的数据,说明Flink在处理乱序大实时数据的时候,更有优势。
容错机制	SparkStreaming的容错机制是基于RDD的容错机制，会将经常用的RDD或者对宽依赖加Checkpoint。利用SparkStreaming的direct方式与Kafka可以保证数据输入源的，处理过程，输出过程符合exactly once。	Flink 则使用两阶段提交协议来保证exactly once。
吞吐量与延迟	spark是基于微批的,而且流水线优化做的很好,所以说他的吞入量是最大的,但是付出了延迟的代价,它的延迟是秒级。	而Flink是基于事件的,消息逐条处理,而且他的容错机制很轻量级,所以他能在兼顾高吞吐量的同时又有很低的延迟,它的延迟能够达到毫秒级。

Flink原理介绍先写到这里了，更多关于Flink的知识点，请您耐心等待，当然也可以先自行去看官方文档：https://nightlies.apache.org/flink/。

大数据Hadoop之——实时计算流计算引擎Flink（Flink环境部署）

大数据系统相关栏目本月热门文章