yarn-session模式提交flink任务步骤

yarn-sessionsh(开辟资源) + flink run(提交任务)

1在yarn上启动一个Flink会话，node1上执行以下命令

/export/server/flink/bin/yarn-sessionsh -n 2 -tm 800 -s 1 -d

说明:

申请2个CPU、1600M内存

-n 表示申请2个容器，这里指的就是多少个taskmanager

-tm 表示每个TaskManager的内存大小

-s 表示每个TaskManager的slots数量

-d 表示以后台程序方式运行

注意:

该警告不用管

WARN orgapachehadoophdfsDFSClient - Caught exception

javalangInterruptedException

>

Flink 是一个流处理框架，支持流处理和批处理，特点是流处理有限，可容错，可扩展，高吞吐，低延迟。

流处理是处理一条，立马下一个节点会从缓存中取出，在下一个节点进行计算

批处理是只有处理一批完成后，才会经过网络传输到下一个节点

流处理的优点是低延迟 批处理的优点是高吞吐

flink同时支持两种，flink的网络传输是设计固定的缓存块为单位，用户可以设置缓存块的超时值来决定换存块什么时候进行传输。 数据大于0 进行处理就是流式处理。

如果设置为无限大就是批处理模型。

Flink 集群包括 JobManager 和 TaskManager

JobManager 主要负责调度 Job 并协调 Task 做 checkpoint，职责上很像 Storm 的 Nimbus。从 Client 处接收到 Job 和 JAR 包 等资源后，会生成优化后的执行计划，并以 Task 的单元调度到各个 TaskManager 去执行。

TaskManager 在启动的时候就设置好了槽位数（Slot），每个 slot 能启动一个 Task，Task 为线程。从 JobManager 处接收需要 部署的 Task，部署启动后，与自己的上游建立 Netty 连接，接收数据并处理。

flink on yarn 是由client 提交 app到 RM 上， 然后RM 分配一个 AppMaster负责运行 Flink JobManager 和 Yarn AppMaster, 然后 AppMaster 分配 容器去运行 Flink TaskManger

SparkStreaming 是将流处理分成微批处理的作业， 最后的处理引擎是spark job

Spark Streaming把实时输入数据流以时间片Δt （如1秒）为单位切分成块，Spark Streaming会把每块数据作为一个RDD，并使用RDD *** 作处理每一小块数据。每个块都会生成一个Spark Job处理，然后分批次提交job到集群中去运行，运行每个 job的过程和真正的spark 任务没有任何区别。

JobScheduler, 负责 Job的调度通过定时器每隔一段时间根据Dstream的依赖关系生一个一个DAG图

ReceiverTracker负责数据的接收，管理和分配

ReceiverTracker在启动Receiver的时候他有ReceiverSupervisor,其实现是ReceiverSupervisorImpl, ReceiverSupervisor本身启 动的时候会启动Receiver，Receiver不断的接收数据，通过BlockGenerator将数据转换成Block。定时器会不断的把Block数据通会不断的把Block数据通过BlockManager或者WAL进行存储，数据存储之后ReceiverSupervisorlmpl会把存储后的数据的元数据Metadate汇报给ReceiverTracker，其实是汇报给ReceiverTracker中的RPC实体ReceiverTrackerEndpoin

spark on yarn 的cluster模式， Spark client 向RM提交job请求, RM会分配一个 AppMaster, driver 和 运行在AppMAster节点里， AM然后把Receiver作为一个Task提交给Spark Executor 节点， Receive启动接受数据，生成数据块，并通知Spark Appmaster, AM会根据数据块生成相应的Job, 并把Job 提交给空闲的 Executor 去执行。

1：需要关注流数据是否需要进行状态管理

2：At-least-once或者Exectly-once消息投递模式是否有特殊要求

3：对于小型独立的项目，并且需要低延迟的场景，建议使用storm

4：如果你的项目已经使用了spark，并且秒级别的实时处理可以满足需求的话，建议使用sparkStreaming

5：要求消息投递语义为 Exactly Once 的场景；数据量较大，要求高吞吐低延迟的场景；需要进行状态管理或窗口统计的场景，建议使用flink

Flink 提供的Api右 DataStream 和 DataSet ，他们都是不可变的数据集合，不可以增加删除中的元素， 通过 Source 创建 DataStream 和 DataSet

在创建运行时有：

Flink的每一个Operator称为一个任务， Operator 的每一个实例称为子任务，每一个任务在JVM线程中执行。可以将多个子任务链接成一个任务，减少上下文切换的开销，降低延迟。

source 和 算子map 如果是 one by one 的关系，他们的数据交换可以通过缓存而不是网络通信

TaskManager 为控制执行任务的数量，将计算资源划分多个slot,每个slot独享计算资源，这种静态分配利于任务资源隔离。

同一个任务可以共享一个slot, 不同作业不可以。

这里因为 Source 和 Map 并行度都是4 采用直连方式，他们的数据通信采用缓存形式

所以一共需要两个TaskManager source,Map 一个，reduce一个， 每个TaskManager 要3个slot

JobManager 将 JobGraph 部署 ExecutionGraph

设置的并行度，可以让一个ExecJobVertex 对应 多个并行的ExecVertex 实例。

Flink通过状态机管理 ExecGraph的作业执行进度。

Flink 将对象序列化为固定数量的预先分配的内存段，而不是直接把对象放在堆内存上。

Flink TaskManager 是由几个内部组件组成的：actor 系统（负责与 Flink master 协调）、IOManager（负责将数据溢出到磁盘并将其读取回来）、MemoryManager（负责协调内存使用。

数据源：

Sink:

时间：

处理时间：取自Operator的机器系统时间

事件时间： 由数据源产生

进入时间： 被Source节点观察时的系统时间

如果数据源没有自己正确创建水印，程序必须自己生成水印来确保基于事件的时间窗口可以正常工作。。

DataStream 提供了 周期性水印，间歇式水印，和递增式水印

Flink是依赖内存计算，计算过程中内存不够对Flink的执行效率影响很大。可以通过监控GC（Garbage Collection），评估内存使用及剩余情况来判断内存是否变成性能瓶颈，并根据情况优化。

监控节点进程的YARN的Container GC日志，如果频繁出现Full GC，需要优化GC。

GC的配置：在客户端的“conf/flink-confyaml”配置文件中，在“envjavaopts”配置项中添加参数：“

此处默认已经添加GC日志。

任务的并行度可以通过以下四种层次（按优先级从高到低排列）指定，用户可以根据实际的内存、CPU、数据以及应用程序逻辑的情况调整并行度参数。

您可以根据实际任务数量的多少，为JobManager设置一个合适的内存。

•在使用yarn-session命令时，添加“-jm MEM”参数设置内存。

•在使用yarn-cluster命令时，添加“-yjm MEM”参数设置内存。

每个TaskManager每个核同时能跑一个task，所以增加了TaskManager的个数相当于增大了任务的并发度。在资源充足的情况下，可以相应增加TaskManager的个数，以提高运行效率。

•在使用yarn-session命令时，添加“-n NUM”参数设置TaskManager个数。

•在使用yarn-cluster命令时，添加“-yn NUM”参数设置TaskManager个数。

每个TaskManager多个核同时能跑多个task，相当于增大了任务的并发度。但是由于所有核共用TaskManager的内存，所以要在内存和核数之间做好平衡。

•在使用yarn-session命令时，添加“-s NUM”参数设置SLOT数。

•在使用yarn-cluster命令时，添加“-ys NUM”参数设置SLOT数。

TaskManager的内存主要用于任务执行、通信等。当一个任务很大的时候，可能需要较多资源，因而内存也可以做相应的增加。

•将在使用yarn-sesion命令时，添加“-tm MEM”参数设置内存。

•将在使用yarn-cluster命令时，添加“-ytm MEM”参数设置内存。

周末了，不想搞长篇大论，就写写这样的流水账吧。

Flink的常见异常众多，不可能面面俱到，所以想到哪儿写到哪儿，有漏掉的之后再补充。

这不是个显式错误，但是JDK版本过低很有可能会导致Flink作业出现各种莫名其妙的问题，因此在生产环境中建议采用JDK 8的较高update（我们使用的是181）。

该信息不甚准确，因为绝大多数情况下都不是JAR包本身有毛病，而是在作业提交过程中出现异常退出了。因此需要查看本次提交产生的客户端日志（默认位于$FLINK_HOME/logs目录下），再根据其中的信息定位并解决问题。

一般都是因为用户依赖第三方包的版本与Flink框架依赖的版本有冲突导致。如果是采用Maven做项目管理的话，可参照我之前写的 这篇文章 来解决冲突。

就是字面意思，YARN集群内没有足够的资源启动Flink作业。检查一下当前YARN集群的状态、正在运行的YARN App以及Flink作业所处的队列，释放一些资源或者加入新的资源。

slot分配请求超时，是因为TaskManager申请资源时无法正常获得，按照上一条的思路检查即可。

TaskManager的Container因为使用资源超限被kill掉了。首先需要保证每个slot分配到的内存量足够，特殊情况下可以手动配置SlotSharingGroup来减少单个slot中共享Task的数量。如果资源没问题，那么多半就是程序内部发生了内存泄露。建议仔细查看TaskManager日志，并按处理JVM OOM问题的常规 *** 作来排查。

TaskManager心跳超时。有可能是TaskManager已经失败，如果没有失败，那么有可能是因为网络不好导致JobManager没能收到心跳信号，或者TaskManager忙于GC，无法发送心跳信号。JobManager会重启心跳超时的TaskManager，如果频繁出现此异常，应该通过日志进一步定位问题所在。

Flink on YARN的其他问题，还可以参考 这篇 ，非常有帮助。

该异常几乎都是由于程序业务逻辑有误，或者数据流里存在未处理好的脏数据导致的，继续向下追溯异常栈一般就可以看到具体的出错原因，比较常见的如POJO内有空字段，或者抽取事件时间的时间戳为null等。

很多童鞋拿着这两条异常信息来求助，但实际上它们只是表示BufferPool、MemoryManager这些Flink运行时组件被销毁，亦即作业已经失败。具体的原因多种多样，根据经验，一般是上一条描述的情况居多（即Could not forward element to next operator错误会伴随出现），其次是JDK版本问题。具体情况还是要根据TaskManager日志具体分析。

Akka超时导致，一般有两种原因：一是集群负载比较大或者网络比较拥塞，二是业务逻辑同步调用耗时的外部服务。如果负载或网络问题无法彻底缓解，需考虑调大 akkaasktimeout 参数的值（默认只有10秒）；另外，调用外部服务时尽量异步 *** 作（Async I/O）。

这个异常我们应该都不陌生，首先检查系统 ulimit -n 的文件描述符限制，再注意检查程序内是否有资源（如各种连接池的连接）未及时释放。值得注意的是，Flink使用RocksDB状态后端也有可能会抛出这个异常，此时需修改flink-confyaml中的 statebackendrocksdbfilesopen 参数，如果不限制，可以改为-1。

关于文件描述符的一些有趣知识，可以参见之前我写的 这一篇 。

在Flink内使用Java Lambda表达式时，由于类型擦除造成的副作用（详情见 这篇文章 ），注意调用returns()方法指定被擦除的类型。

在当前检查点还未做完时，收到了更新的检查点的barrier，表示当前检查点不再需要而被取消掉，一般不需要特殊处理。

首先应检查 CheckpointConfigsetCheckpointTimeout() 方法设定的检查点超时，如果设的太短，适当改长一点。另外就是考虑发生了反压或数据倾斜，或者barrier对齐太慢。具体思路不再赘述，看官可以参考 这篇文章 ，非常详细。

我们知道Flink的状态是按key组织并保存的，如果程序逻辑内改了keyBy()逻辑或者key的序列化逻辑，就会导致检查点/保存点的数据无法正确恢复。所以如果必须要改key相关的东西，就弃用之前的状态数据吧。

在19之前的Flink版本中，如果我们使用RocksDB状态后端，并且更改了自用MapState的schema，恢复作业时会抛出此异常，表示不支持更改schema。这个问题已经在 FLINK-11947 解决，升级版本即可。

就酱吧，民那晚安（不是

以上就是关于yarn-session模式提交flink任务步骤全部的内容，包括:yarn-session模式提交flink任务步骤、flink实战教程-集群的部署、Flink 原理详解等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！