Question

Flink 中的类加载概述

运行 Flink 应用程序时，JVM 将随着时间的推移加载各种类。这些类可以分为两个域：

• 在 Java 类路径 ：这是 Java 类路径中常见的，它包括了 JDK 库，所有代码 Flink 的 /lib 文件夹（阿帕奇 Flink 及其核心依赖的类）。
• 该 动态用户代码 ：这些是包含在动态提交的作业，的 JAR 文件的所有类（通过 REST，CLI，web 用户界面）。它们按工作动态加载（和卸载）。

哪些类属于哪个域取决于运行 Apache Flink 的特定设置。作为一般规则，无论何时首先启动 Flink 进程和提交作业，都会动态加载作业的类。如果 Flink 进程与作业/应用程序一起启动，或者应用程序生成 Flink 组件（JobManager，TaskManager 等），则所有类都在 Java 类路径中。

以下是有关不同部署模式的更多详细信息：

独立会话

将 Flink 集群作为独立会话启动时，JobManagers 和 TaskManagers 将使用 Java 类路径中的 Flink 框架类启动。来自会话（通过 REST / CLI）提交的所有作业/应用程序的类都是 动态 加载的。

Docker / Kubernetes Sessions

Docker / Kubernetes 设置首先启动一组 JobManagers / TaskManagers，然后通过 REST 或 CLI 提交作业/应用程序，就像独立会话一样：Flink 的代码在 Java 类路径中，作业的代码是动态加载的。

YARN

YARN 类加载在单个作业部署和会话之间有所不同：

• 当直接向 YARN（via bin/flink run -m yarn-cluster ... ）提交 Flink 作业/应用程序时，将为该作业启动专用 TaskManagers 和 JobManagers。这些 JVM 在 Java 类路径中同时具有 Flink 框架类和用户代码类。这意味着在这种情况下 不 涉及 动态类加载 。
• 在启动 YARN 会话时，JobManagers 和 TaskManagers 将使用类路径中的 Flink 框架类启动。针对会话提交的所有作业的类都是动态加载的。

Mesos

遵循 此文档的 Mesos 设置目前非常类似于 YARN 会话：TaskManager 和 JobManager 进程使用 Java 类路径中的 Flink 框架类启动，作业类在提交作业时动态加载。

反向类加载和类加载器分辨率顺序

在涉及动态类加载（会话）的设置中，通常有两个 ClassLoader 的层次结构：（1）Java 应用程序类加载器 ，它具有类路径中的所有类，以及（2）动态 用户代码类加载器 。用于从用户代码 jar 加载类。用户代码 ClassLoader 将应用程序类加载器作为其父代。案例。

默认情况下，Flink 反转类加载顺序，这意味着它首先查看用户代码类加载器，并且只查看父类（应用程序类加载器），如果该类不是动态加载的用户代码的一部分。

反向类加载的好处是作业可以使用与 Flink 核心本身不同的库版本，这在库的不同版本不兼容时非常有用。该机制有助于避免常见的依赖冲突错误，如 IllegalAccessError 或 NoSuchMethodError 。代码的不同部分只是具有类的单独副本（Flink 的核心或其中一个依赖项可以使用与用户代码不同的副本）。在大多数情况下，这种方法运行良好，无需用户进行其他配置。

但是，有些情况下反向类加载会导致问题（参见下文“X 不能转换为 X”）。您可以通过在 Flink 配置中通过 classloader.resolve-order 配置 ClassLoader 解析顺序 parent-first （从 Flink 的默认设置 child-first ）恢复到 Java 默认模式 。

请注意，某些类总是以 父对象的 方式解析（ 首先 通过父 ClassLoader），因为它们在 Flink 的核心和用户代码或面向 API 的用户代码之间共享。这些类的包通过 classloader.parent-first-patterns-default 和 classloader.parent-first-patterns-additional 配置 。要添加要以 父级优先 加载的新软件包，请设置 classloader.parent-first-patterns-additional config 选项。

避免动态类加载

所有组件（JobManger，TaskManager，Client，ApplicationMaster，...）在启动时记录其类路径设置。它们可以在日志开头的环境信息中找到。

当运行 Flink JobManager 和 TaskManagers 专用于某个特定作业的设置时，可以将 JAR 文件直接放入该 /lib 文件夹中，以确保它们是类路径的一部分而不是动态加载。

它通常用于将作业的 JAR 文件放入 /lib 目录中。JAR 将成为类路径（ AppClassLoader ）和动态类加载器（ FlinkUserCodeClassLoader ）的一部分。因为 AppClassLoader 是 FlinkUserCodeClassLoader 的父级（并且默认情况下 Java 加载父级优先），所以这应该导致只加载一次类。

对于无法将作业的 JAR 文件放入 /lib 文件夹的设置（例如，因为安装程序是多个作业使用的会话），可能仍然可以将公共库放入该 /lib 文件夹，并避免为这些文件夹加载动态类。

在作业中手动加载类

在某些情况下，转换函数，源或接收器需要手动加载类（通过反射动态加载）。要做到这一点，它需要可以访问作业类的类加载器。

在这种情况下，函数（或源或接收器）可以成为 RichFunction （例如 RichMapFunction 或 RichWindowFunction ）并通过访问用户代码类加载器 getRuntimeContext().getUserCodeClassLoader() 。

X 不能转换为 X 异常

在使用动态类加载的设置中，您可能会在样式中看到异常 com.foo.X cannot be cast to com.foo.X 。这意味着该类的多个版本 com.foo.X 已由不同的类加载器加载，并且尝试将该类的类型彼此分配。

一个常见原因是库与 Flink 的 反向类加载 方法不兼容。您可以关闭反向类加载以验证这一点（ classloader.resolve-order: parent-first 在 Flink 配置中设置）或从反向类加载中排除库（ classloader.parent-first-patterns-additional 在 Flink 配置中设置）。

另一个原因可能是缓存的对象实例，如某些库（如 Apache Avro ）或实习对象（例如通过 Guava 的 Interners）生成的。这里的解决方案是要么没有任何动态类加载的设置，要么确保相应的库完全是动态加载代码的一部分。后者意味着库不能添加到 Flink 的 /lib 文件夹中，但必须是应用程序的 fat-jar / uber-jar 的一部分

卸载动态加载的类

涉及动态类加载（会话）的所有方案都依赖于再次 卸载的 类。类卸载意味着垃圾收集器发现类中没有对象存在且更多，因此删除了类（代码，静态变量，元数据等）。

每当 TaskManager 启动（或重新启动）任务时，它都会加载该特定任务的代码。除非可以卸载类，否则这将成为内存泄漏，因为加载了新版本的类，并且加载的类的总数随时间累积。这通常通过 OutOfMemoryError：Metaspace 表现出来。

类泄漏的常见原因和建议的修复：

• 延续线程 ：确保应用程序函数/ sources / sinks 关闭所有线程。延迟线程本身会花费资源，另外通常还包含对（用户代码）对象的引用，从而防止垃圾收集和卸载类。
• Interners ：避免在超出函数/源/接收器生命周期的特殊结构中缓存对象。示例是 Guava 的 interner，或序列化程序中 Avro 的类/对象缓存。

使用 maven-shade-plugin 解决与 Flink 的依赖冲突。

A 到解决从应用程序开发者方面的依赖性冲突的方法是避免通过暴露的依赖 阴影他们离开 。

Apache Maven 提供 maven-shade-plugin ，它允许 在 编译 后 更改类的包（因此您编写的代码不受着色影响）。例如，如果你 com.amazonaws 的用户代码 jar 中有来自 aws sdk 的 org.myorg.shaded.com.amazonaws 软件包，那么 shade 插件会将它们重定位到软件包中，这样你的代码就会调用你的 aws sdk 版本。

本文档页面介绍了 使用 shade 插件重定位类 。

请注意，大多数 Flink 的依赖，如 guava ， netty ， jackson ，等被 Flink 的维护者阴影了，所以用户通常不必担心。