Hi Everyone, I am trying to integrate apache kafka, apache Spark Streaming for data processing. I have included the code, error and pom.xml below. I need help error Spark Streaming Started now... Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/spark/streaming/api/java/JavaStreamingContext at org.example.SparkKafkaStream.main(SparkKafkaStream.java:22) Caused by: java.lang.ClassNotFoundException: org.apache.spark.streaming.api.java.JavaStreamingContext at java.net.URLClassLoader.findClass(URLClassLoader.java:382) at java.lang.ClassLoader.loadClass(ClassLoader.java:418) at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:355) at java.lang.ClassLoader.loadClass(ClassLoader.java:351) ... 1 more Process finished with exit code 1. pom.xml <? xml version ="1.0" encoding ="UTF-8" ?> <project xmlns ="http://maven.apache.org/POM/4.0.0" xmlns: xsi ="http://www.w3.org/2001/XMLSchema-instance" xsi :schemaLocation ="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" > <modelVersion> 4.0.0 </modelVersion> <groupId> org.example </groupId> <artifactId> sparktwitter </artifactId> <version> 1.0-SNAPSHOT </version> <name> sparktwitter </name> <!-- FIXME change it to the project's website --> <url> http://www.example.com </url> <properties> <project.build.sourceEncoding> UTF-8 </project.build.sourceEncoding> <maven.compiler.source> 1.7 </maven.compiler.source> <maven.compiler.target> 1.7 </maven.compiler.target> </properties> <dependencies> <dependency> <groupId> junit </groupId> <artifactId> junit </artifactId> <version> 4.11 </version> <scope> test </scope> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming --> <dependency> <groupId> org.apache.spark </groupId> <artifactId> spark-streaming_2.12 </artifactId> <version> 2.4.7 </version> <scope> provided </scope> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-twitter --> <dependency> <groupId> org.apache.spark </groupId> <artifactId> spark-streaming-twitter_2.11 </artifactId> <version> 1.6.3 </version> </dependency> </dependencies> <build> <pluginManagement> <!-- lock down plugins versions to avoid using Maven defaults (may be moved to parent pom) --> <plugins> <!-- clean lifecycle, see https://maven.apache.org/ref/current/maven-core/lifecycles.html#clean_Lifecycle --> <plugin> <artifactId> maven-clean-plugin </artifactId> <version> 3.1.0 </version> </plugin> <!-- default lifecycle, jar packaging: see https://maven.apache.org/ref/current/maven-core/default-bindings.html#Plugin_bindings_for_jar_packaging --> <plugin> <artifactId> maven-resources-plugin </artifactId> <version> 3.0.2 </version> </plugin> <plugin> <artifactId> maven-compiler-plugin </artifactId> <version> 3.8.0 </version> </plugin> <plugin> <artifactId> maven-surefire-plugin </artifactId> <version> 2.22.1 </version> </plugin> <plugin> <artifactId> maven-jar-plugin </artifactId> <version> 3.0.2 </version> </plugin> <plugin> <artifactId> maven-install-plugin </artifactId> <version> 2.5.2 </version> </plugin> <plugin> <artifactId> maven-deploy-plugin </artifactId> <version> 2.8.2 </version> </plugin> <!-- site lifecycle, see https://maven.apache.org/ref/current/maven-core/lifecycles.html#site_Lifecycle --> <plugin> <artifactId> maven-site-plugin </artifactId> <version> 3.7.1 </version> </plugin> <plugin> <artifactId> maven-project-info-reports-plugin </artifactId> <version> 3.0.0 </version> </plugin> </plugins> </pluginManagement> <plugins> <plugin> <groupId> org.apache.maven.plugins </groupId> <artifactId> maven-compiler-plugin </artifactId> <configuration> <source> 8 </source> <target> 8 </target> </configuration> </plugin> </plugins> </build> </project> Code. package org.example ; import java.io.IOException ; import java.util.* ; import kafka.serializer.StringDecoder ; import org.apache.spark.* ; import org.apache.spark.streaming.* ; import org.apache.spark.streaming.api.java.* ; import org.apache.spark.streaming.kafka.KafkaUtils ; public class SparkKafkaStream { public static void main (String[] args) throws IOException { System. out .println( "Spark Streaming Started now..." ) ; // Create a local StreamingContext with two working thread and batch interval of 1 second final SparkConf conf = new SparkConf().setMaster( "local[2]" ).setAppName( "Data Processing" ) ; //JavaSparkContext sc = new JavaSparkContext(conf); final JavaStreamingContext ssc = new JavaStreamingContext(conf , Durations. seconds ( 1 )) ; Map<String , String> kafkaParams = new HashMap<>() ; kafkaParams.put( "metadata.broker.list" , "localhost:9092" ) ; Set<String> topics = Collections. singleton ( "Demo" ) ; JavaPairInputDStream<String , String> directKafkaStream = KafkaUtils. createDirectStream ( ssc , String. class, String. class, StringDecoder. class, StringDecoder. class, kafkaParams , topics ) ; final List<String> allRecord = new ArrayList<String>() ; final String COMMA = "," ; directKafkaStream.foreachRDD(rdd -> { System. out .println( "New data arrived " + rdd.partitions().size() + " Partitions and " + rdd.count() + " Records" ) ; if (rdd.count() > 0 ) { rdd.collect().forEach(rawRecord -> { System. out .println(rawRecord) ; System. out .println( "***************************************" ) ; System. out .println(rawRecord. _2 ) ; String record = rawRecord._2() ; StringTokenizer st = new StringTokenizer(record , "," ) ; }) ; } }) ; ssc.start() ; try { ssc.awaitTermination() ; } catch (InterruptedException e) { e.printStackTrace() ; } } }
... View more
