在 flink yarn 集群上通过 maven 运行包含配置的 Java Jar

Question

我正在一个 maven/java 项目中使用 flink，需要在创建的 jar 内部包含我的配置。

因此，我在我的pom文件中添加了以下内容。这包括我在 jar 中的所有 yml 配置(位于 src/main/resources 文件夹中(，我在执行时将它的名称作为参数传递。

<resources>
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.yml</include>
</includes>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.4.3</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
<finalName>${project.artifactId}-${project.version}</finalName>
<shadedArtifactAttached>true</shadedArtifactAttached>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.exmaple.MyApplication</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>

下面的主类代码接收一个参数，根据它我决定从资源中选择什么配置，读取(使用 snakeyaml(并使用。

public static void main(String[] args) throws Exception {
final ParameterTool parameterTool = ParameterTool.fromArgs(args);
ClassLoader classLoader = MyApplication.class.getClassLoader();
Yaml yaml = new Yaml();
String filename = parameterTool.getRequired("configFilename");
InputStream in  = classLoader.getSystemResourceAsStream(filename);
MyConfigClass = yaml.loadAs(in, MyConfigClass.class);
...
}

MVN 全新安装创建"我的阴影罐.jar">

我使用命令执行

java -jar /path/to/my-shaded-jar.jar --configFilename filename

如果我与他人共享 jar，它可以在多个系统上工作。

但是，当我尝试使用以下命令在Hadoop上的纱线集群中运行相同的jar时，我遇到了问题：-

HADOOP_CLASSPATH=`hadoop classpath` HADOOP_CONF_DIR=/etc/hadoop/conf ./flink-1.6.2/bin/flink run -m yarn-cluster -yd -yn 5 -ys 30 -yjm 10240 -ytm 10240 -yst -ynm some-job-name -yqu queue-name ./my-shaded-jar.jar --configFilename filename

我收到以下错误：

------------------------------------------------------------
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error.
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:546)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)
at org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:83)
at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:78)
at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:120)
at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:238)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:216)
at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1053)
at org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1129)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1754)
at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1129)
Caused by: org.yaml.snakeyaml.error.YAMLException: java.io.IOException: Stream closed
at org.yaml.snakeyaml.reader.StreamReader.update(StreamReader.java:200)
at org.yaml.snakeyaml.reader.StreamReader.<init>(StreamReader.java:60)
at org.yaml.snakeyaml.Yaml.loadAs(Yaml.java:444)
at com.example.MyApplication.main(MyApplication.java:53)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)
... 13 more
Caused by: java.io.IOException: Stream closed
at java.io.PushbackInputStream.ensureOpen(PushbackInputStream.java:74)
at java.io.PushbackInputStream.read(PushbackInputStream.java:166)
at org.yaml.snakeyaml.reader.UnicodeReader.init(UnicodeReader.java:90)
at org.yaml.snakeyaml.reader.UnicodeReader.read(UnicodeReader.java:122)
at java.io.Reader.read(Reader.java:140)
at org.yaml.snakeyaml.reader.StreamReader.update(StreamReader.java:184)

为什么我的解决方案可以在任何普通的 linux/mac 系统上工作，但是在 yarn 集群上使用 flink run 命令运行时，具有相同参数的相同 jar 失败。 我们通常执行罐子的方式和纱线的方式之间有区别吗？

任何帮助表示赞赏。

Answer 1

将classLoader.getSystemResourceAsStream(filename)替换为classLoader.getResourceAsStream(filename)。

1. java.lang.ClassLoader#getSystemResourceAsStream通过系统类装入器查找资源，该装入器通常用于启动应用程序。

2. java.lang.ClassLoader#getResourceAsStream将首先搜索父类装入器。如果失败，它将调用当前类装入器的findResource。

为了避免依赖冲突，Flink 应用中的类被划分为两个域 [1]，这些域也适用于 Flink 客户端，例如CliFrontend.

Java Classpath包括 Apache Flink 的类及其核心依赖项。动态用户代码包括用户jar 的类(和资源(。

所以为了找到打包在你的jar文件中的"配置文件"，我们应该使用用户代码类加载器(你可以在org.apache.flink.client.program.PackagedProgram中找到userCodeClassLoader的详细信息(，而不是系统类加载器。

1. https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/debugging_classloading.html