在使用 Flink 的生产实践中，我们发现采用 Flink on YARN 的部署方式时，将大量用户依赖的 JAR 包和其他文件上传到对应的容器中是从用户发送部署请求到应用实际运行起来的重要耗时操作。其中，不少文件例如 Flink 框架本身的依赖对于每个应用来说都是一样的，如果能采用一个统一的缓存将会显著减少所需上传的文件大小；此外，许多用户依赖的 JAR 包只在运行时需要，而这些依赖可能本身是存储在 YARN 集群上的，在提交之前先将依赖拉取下来再上传上去将会带来额外无谓的开销。本文首先介绍 YARN 上资源本地化的基础知识，随后介绍利用这一技术减少 Flink 在 YARN 上部署时间的几个具体工作

YARN 上的资源本地化

关于 YARN 上的资源本地化，Cloudera 的这篇文章是个很好的介绍，本节内容基本是链接文章的一个翻译和简单的演绎。

YARN 上的资源本地化主要涉及以下几个概念

• 本地化（Localization） - 本地化是一个将远端资源下载或复制到本地文件系统的过程，通过本地化的过程，在访问文件时就不再需要每次都从远端拉资源，只需要读取本地的资源
• 本地资源（LocalResource） - 本地资源代表运行一个容器所需的文件或者库。NodeManager 会负责在启动容器之前先本地化所需的资源。对于每个本地资源，应用可以指定资源的以下属性
• URL - 下载资源所需的远端地址
• 大小（Size） - 本地资源的大小，以字节（Byte）为单位
• 资源在远端文件系统上创建时的时间戳
• 本地资源类型（LocalResourceType） - 指定 NodeManager 本地化的资源的类型，包括 FILE、ARCHIVE 和 PATTERN
• 模式（Pattern） - 解析文件的模式（仅当本地资源类型为 PATTERN 是有效）
• 本地资源可见性（LocalResourceVisibility） - 指定 NodeManager 本地化的资源的可见性，包括 PUBLIC、PRIVATE 和 APPLICATION
• 资源本地化服务（ResourceLocalizationService） - NodeManager 中负责本地化的服务
• 删除服务（DeletionService） - NodeManager 中负责接受请求并删除本地路径的服务
• 本地化器（Localizer） - 实际进行本地化的线程或进程。本地化器有两种类型，为 PUBLIC 资源进行本地化的 PublicLocalizer 以及为 PRIVATE 和 APPLICATION 资源进行本地化的 ContainerLocalizers
• 本地缓存（LocalCache） - NodeManager 管理并维护了好几种本地缓存，这些缓存包括了所有下载下来的文件。其中的资源由被拉取时指定的远端 URL 唯一确定

YARN 上的资源本地化的运行过程根据不同的本地资源类型有所不同，我们在 Flink 的优化中采用不同的本地资源类型进行不同方向的优化

PUBLIC 资源的本地化相关流程如下

• PublicLocalizers 运行在 NodeManager 的地址空间中
• PublicLocalizers 线程的数量由配置项 yarn.nodemanager.localizer.fetch.thread-count 指定，这是下载 PUBLIC 资源是的最大并行度
• 本地化 PULBIC 资源时，本地化器会校验所有需要的资源在远端文件系统上有正确的权限，不满足这个条件的资源将被拒绝本地化
• PublicLocalizer 采用 ContainerLaunchContext 中配置的证书拉取资源以保证访问远端文件系统时的安全性需求

PRIVATE/APPLICATON 资源的本地化相关流程如下

• ContainerLocalizer 运行在一个隔离进程当中，即和 NodeManager 运行在不同线程当中
• 每个 ContainerLocalizer 进程都由 NodeManager 中的一个线程管理，称之为 LocalizerRunner。每个容器在启动时，如果发现有任何还没下载的资源，就会启动一个 LocalizerRunner
• 同时，本地化的过程还会包括一个称为 LocalResourcesTracker 的对象。这个对象对应一个用户或者一个应用，会记录所有对应用户或应用的本地资源
• 当容器第一次请求 PRIVATE/APPLICATION 资源时，如果资源在 LocalResourcesTracker 上没找到（或者出于 INITIALIZED 状态），这个资源就会被加入到 pending-resources 列表上。根据是否需要下载资源，LocalizerRunner 将按需创建
• 具体本地化的过程如下
• 在容器启动时，ContainerLocalizer 建立起和 NodeManager 之间的心跳
• 每次心跳，LocalizerRunner 选择向 ContainerLocalizer 请求本地化一份资源或者通知它关闭；ContainerLocalizer 向 LocalizerRunner 汇报资源下载的状态
• 如果资源下载失败，那么这个资源将从 LocalResourcesTracker 中被移除并最终被标记为本地化失败。一旦下载失败，LocalizerRunners 将关闭 ContainerLocalizer 后退出
• 如果资源下载成功，LocalizerRunner 会向 ContainerLocalizer 不断地提出新的下载请求，直到所有资源都成功下载

资源本地化之后，根据本地资源类型的不同将会被下载到不同的目录

• PUBLIC <local-dir>/filecache
• PRIVATE <local-dir>/usercache//filecache
• APPLICATION <local-dir>/usercache//appcache/<app-id>/

Flink 在 YARN 上的部署开销的优化

支持直接指定远端资源为作业依赖

某些运行时依赖本身就在 YARN 集群上作为公共资源存在，我们可以在部署之前通过设置 YARN 的 LocalResource 指定远端 URL 等属性，在应用启动时直接从 YARN 上拉取资源。这样我们可以省去目前 Flink 流程中先将资源下载到本地之后再上传到 YARN 上的临时目录的开销，同时减少了退出时清理临时目录下资源的开销

对于某些公共依赖，我们甚至可以在此之上将资源指定为 PUBLIC 类型的。这样，YARN 在本地化的时候会将资源保存到 <local-dir>/filecache 目录下，对于这个 NodeManager 上的所有 Application 都能共享这份材料。此后，对于新的应用，我们仍然指定本地化资源，但是从 YARN 的角度则只会在 appcache 中简单地建立一个到 filecache 的软连接。通过这种方式我们能进一步减少不同 Application 下载公共依赖的开销

支持指定远端的 Flink 框架 JAR 包

思路和上面的类似，只是 Flink 框架 JAR 包作为必选项需要单独处理，同时有 classpath 组装的问题，不好直接放进作业依赖的加载里。目前 Flink 只支持本地文件系统的 Flink 框架 JAR 包，这是一个实现上没啥道理的限制。直接解开这个限制并在 Flink 框架 JAR 为远端资源时（通过 scheme 判断）指定 LocalResource 而不是拉取并上传

其他和 YARN 本地化相关的优化，包括目前发布的 Flink 版本会在 TM 上部署一份专门的配置文件，这是由于 TM 的配置根据运行时信息和 JM 上的配置文件有出入。但是这部分的差别并不大，社区近期的修改将差别编码到 dynamicPropoties 中，即一个字符串，随容器启动。这样可以减少部署配置文件的开销