Flink 应用程序在 Java 中抛出 Class Not Found 异常

Question

我有一个带有 Yarn 的 Flink 集群，使用 flink-quickstart-java Archetype 构建一个演示项目。使用'mvn clean package -Pbuild-jar'命令构建一个fat-jar，并使用'flink run -m yarn-cluster -yn 2 ./flink-SNAPSHOT-1.0.jar'提交程序，程序抛出以下内容例外：

java.lang.NoClassDefFoundError: java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArrayDeserializer 在 org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.setDeserializer(FlinkKafkaConsumer09.java:290) 在 org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.(FlinkKafkaConsumer09.java:216) 在 org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.(FlinkKafkaConsumer09.java:154) 在 org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010.(FlinkKafkaConsumer010.java:128) 在 org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010.(FlinkKafkaConsumer010.java:112) 在 org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010.(FlinkKafkaConsumer010.java:79) 在stream.TransferKafka.main（TransferKafka.java:19） 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:525) 在 org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:417) 在 org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:395) 在 org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:828) 在 org.apache.flink.client.CliFrontend.run(CliFrontend.java:283) 在 org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1080) 在 org.apache.flink.client.CliFrontend$1.call(CliFrontend.java:1127) 在 org.apache.flink.client.CliFrontend$1.call(CliFrontend.java:1124) 在 java.security.AccessController.doPrivileged（本机方法） 在 javax.security.auth.Subject.doAs(Subject.java:422) 在 org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1781) 在 org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) 在 org.apache.flink.client.CliFrontend.main(CliFrontend.java:1124) 引起：java.lang.ClassNotFoundException：org.apache.kafka.common.serialization.ByteArrayDeserializer 在 java.net.URLClassLoader.findClass(URLClassLoader.java:381) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:424) 在 sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:338) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:357) ... 24 更多

这是我的演示：

public static void main(String[] args) {
 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 Properties props = new Properties();
 props.setProperty("bootstrap.servers", "ip:port");
 props.setProperty("group.id", "NewFlinkTest");
 DataStreamSource < String > stream = env.addSource(new FlinkKafkaConsumer010 < > ("kafka_test", new SimpleStringSchema(), props));
 stream.addSink(new FlinkKafkaProducer010 < > ("kafka_test_out", new SimpleStringSchema(), props));
 try {
  env.execute("Flink Jar Test");
 } catch (Exception e) {
  e.printStackTrace();
 }
}

还有一些版本信息：
FLink 版本：1.4.0

Hadoop 版本：2.7.2

卡夫卡版本：0.10.2.1

JDK 版本：1.8

---

Pom 依赖项

编辑1：

<?xml version="1.0" encoding="UTF-8"?>
<dependencies>
   <!-- Apache Flink dependencies -->
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-core</artifactId>
      <version>${flink.version}</version>
   </dependency>
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-java</artifactId>
      <version>${flink.version}</version>
   </dependency>
   <dependency>
      <!-- This dependency is required to actually execute jobs. It is currently pulled in by           flink-streaming-java, but we explicitly depend on it to safeguard against future changes. -->
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-clients_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
   </dependency>
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
   </dependency>
   <!-- explicitly add a standard logging framework, as Flink does not have     a hard dependency on one specific framework by default -->
   <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>${slf4j.version}</version>
   </dependency>
   <dependency>
      <groupId>log4j</groupId>
      <artifactId>log4j</artifactId>
      <version>${log4j.version}</version>
   </dependency>
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-rabbitmq_2.11</artifactId>
      <version>1.4.0</version>
   </dependency>
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId>
      <version>1.4.0</version>
   </dependency>
</dependencies>

经过一些尝试，我发现代码抛出异常与我打包到我的 uber-jar 中的 jar 不同。我认为主要原因是客户端服务器具有旧版本的 flink-connector-kafka jar，但无论我如何设置配置 yaml 属性 'yarn.per-job-cluster.include-user-jar'，程序总是抛出相同的异常。

---

编辑2：

将 kafka-clients:0.10.2.1 添加到 flink_home/lib/ 后，它可以工作了。但是仍然不知道为什么它不读取 uber jar 中的类文件。

Answer 1

您可能希望将 <scope>provided</scope> 添加到 flink-streaming-scala、flink-clients、link-table-api-scala-bridge 和 flink-table-planner-blink - 这解决了我的问题

Answer 2

首先，您可以通过grep 'ByteArrayDeserializer' ./flink-SNAPSHOT-1.0.jar 验证您的jar 文件中是否存在丢失的类。