NoClassDefFoundError lors de l'exécution du flink avec le connecteur kafka

Question

J'essaie de diffuser des données à partir de kafka à l'aide de flink. Mon code compile sans erreur, mais sur l'exécution, je reçois l'erreur suivante:

Error: A JNI error has occurred, please check your installation and try again 
Exception in thread "main" java.lang.NoClassDefFoundError: 
    org/apache/flink/streaming/util/serialization/DeserializationSchema 
    at java.lang.Class.getDeclaredMethods0(Native Method) 
    at java.lang.Class.privateGetDeclaredMethods(Class.java:2701) 
    at java.lang.Class.privateGetMethodRecursive(Class.java:3048) 
    at java.lang.Class.getMethod0(Class.java:3018) 
    at java.lang.Class.getMethod(Class.java:1784) 
    at sun.launcher.LauncherHelper.validateMainClass(LauncherHelper.java:544) 
    at sun.launcher.LauncherHelper.checkAndLoadMain(LauncherHelper.java:526) 
Caused by: java.lang.ClassNotFoundException: org.apache.flink.streaming.util.serialization.DeserializationSchema 
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424) 
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357) 
    ... 7 more 

Ma POM liste de dépendance est la suivante:

<dependencies> 
     <dependency> 
      <groupId>org.apache.flink</groupId> 
      <artifactId>flink-java</artifactId> 
      <version>1.3.2</version> 
     </dependency> 
     <dependency> 
      <groupId>org.apache.flink</groupId> 
      <artifactId>flink-streaming-core</artifactId> 
      <version>0.9.1</version> 
     </dependency> 
     <dependency> 
      <groupId>org.apache.flink</groupId> 
      <artifactId>flink-clients</artifactId> 
      <version>0.10.2</version> 
     </dependency> 
     <dependency> 
      <groupId>org.apache.flink</groupId> 
      <artifactId>flink-connector-kafka-0.9_2.11</artifactId> 
      <version>1.3.2</version> 
     </dependency> 
     <dependency> 
      <groupId>com.googlecode.json-simple</groupId> 
      <artifactId>json-simple</artifactId> 
      <version>1.1</version> 
     </dependency> 
    </dependencies> 

Le code java que je suis en train de courir souscrit simplement à un kafka sujet appelé 'streamer':

import java.util.Properties; 
import java.util.Arrays; 
import org.apache.flink.api.common.functions.MapFunction; 
import org.apache.flink.api.java.utils.ParameterTool; 
import org.apache.flink.streaming.api.datastream.DataStream; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09; 
import org.apache.flink.streaming.util.serialization.SimpleStringSchema; 
import org.apache.flink.streaming.util.serialization.DeserializationSchema; 

public class StreamConsumer { 
public static void main(String[] args) throws Exception { 
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
     Properties properties = new Properties(); 
     properties.setProperty("bootstrap.servers", "localhost:9092"); 
     properties.setProperty("group.id", "samplegroup"); 

     DataStream<String> messageStream = env.addSource(new FlinkKafkaConsumer09<String>("streamer", new SimpleStringSchema(), properties)); 

     messageStream.rebalance().map(new MapFunction<String, String>() { 
         private static final long serialVersionUID = -6867736771747690202L; 
         @Override 
         public String map(String value) throws Exception { 
           return "Streamed data: " + value; 
         } 
       }).print(); 
     env.execute(); 
} 
} 

informations système:
1. Version Kafka: 0.9.0.1
2. Version Flink: 1.3.2
3. version OpenJDK: 1.8

Bien que j'utilise maven, je ne pense pas que ce soit une question de maven parce que je reçois la même erreur, même lorsque je tente sans Maven. J'ai téléchargé manuellement tous les fichiers .jar nécessaires dans un dossier et spécifié ce chemin de dossier avec l'option -cp lors de la compilation avec javac. J'obtiens la même erreur que ci-dessus pendant l'exécution mais aucune erreur pendant la compilation.

Answer 1

Il semble que le premier problème avec votre pom est que vous utilisez une version différente pour vos importations de flink. Essayez d'utiliser la nouvelle version 1.3.2 pour tous les modules de flink. Cette erreur se produit souvent lorsque vous utilisez des versions incompatibles ou multiples de bibliothèques.

Essayez d'utiliser les dépendances de mise en jachère (en supposant que vous utilisez scala 2.11):

<dependencies> 
    <dependency> 
     <groupId>org.apache.flink</groupId> 
     <artifactId>flink-java</artifactId> 
     <version>1.3.2</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.flink</groupId> 
     <artifactId>flink-streaming-java_2.11</artifactId> 
     <version>1.3.2</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.flink</groupId> 
     <artifactId>flink-clients_2.11</artifactId> 
     <version>1.3.2</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.flink</groupId> 
     <artifactId>flink-connector-kafka-0.9_2.11</artifactId> 
     <version>1.3.2</version> 
    </dependency> 
    <dependency> 
     <groupId>com.googlecode.json-simple</groupId> 
     <artifactId>json-simple</artifactId> 
     <version>1.1</version> 
    </dependency> 
</dependencies> 

Si vous avez toujours le même exemple de code postal de problème, je peux reproduire l'erreur.

Answer 2

NoClassDefFoundError when running flink with kafka connector

Votre code compile et que vous obtenez NoClassDefFoundError, je pense que l'un des votre bibliothèque de dépendance manquante, il est dépendances Compile ou Runtime Dépendances pendant le processus de téléchargement automatique par Maven .pom

donc probablement Ce sont les causes profondes Vous obtenez NoClassDefFoundError

Solution: clean and build

Answer 3

Essayez de supprimer portée si vous l'avez spécifié dans le fichier où certains pom, car elle limite la portée des fichiers de classe pendant l'exécution

Limit Scope in POM file