Bdtjtk

UPDATE: Issue Fixed. Was a versioning issue. Needed to change spark-core and others to version 2.2.0.

I am new to Apache Spark. I am trying to send logs through Kafka and process them using Spark Streaming before passing it to ElasticSearch. However I am getting the following error.

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/spark/sql/execution/streaming/Source$class
at org.apache.spark.sql.kafka010.KafkaSource.(KafkaSource.scala:87)
at org.apache.spark.sql.kafka010.KafkaSourceProvider.createSource(KafkaSourceProvider.scala:152)
at org.apache.spark.sql.execution.datasources.DataSource.createSource(DataSource.scala:240)
at org.apache.spark.sql.streaming.StreamingQueryManager$$anonfun$1.applyOrElse(StreamingQueryManager.scala:245)
at org.apache.spark.sql.streaming.StreamingQueryManager$$anonfun$1.applyOrElse(StreamingQueryManager.scala:241)
at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:279)
at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:279)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:69)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:278)
at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:268)
at org.apache.spark.sql.streaming.StreamingQueryManager.startQuery(StreamingQueryManager.scala:241)
at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:287)
at com.fork.SreamingApp.App.main(App.java:78)
Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.execution.streaming.Source$class
at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:582)
at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:190)
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:499)
... 13 more

Here are my Code files

App.java

public static void main(String args) throws NotSerializableException, InterruptedException {

        App streamingApp = new App();

        SparkSession spark = SparkSession

                  .builder()

                  .appName("Java Spark SQL data sources example")

                  .master("local")

                  .getOrCreate();



        Dataset<Row> ds1 = spark

                  .readStream()

                  .format("kafka")

                  .option("kafka.bootstrap.servers", "localhost:32770")

                  .option("subscribe", "context3")

                  .load();



        StreamingQuery query = ds1.writeStream()

                  .format("console")

                  .start();



        query.awaitTermination();

        System.out.println("DONE");







}

pom.xml

<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>com.fork</groupId>

 <artifactId>SreamingApp</artifactId>

 <version>0.0.1-SNAPSHOT</version>

 <packaging>jar</packaging>



 <name>SreamingApp</name>

 <url>http://maven.apache.org</url>



 <properties>

   <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

 </properties>



 <dependencies>

   <dependency>

     <groupId>com.maxmind.geoip2</groupId>

     <artifactId>geoip2</artifactId>

     <version>2.12.0</version>

   </dependency>





   <dependency>

     <groupId>junit</groupId>

     <artifactId>junit</artifactId>

     <version>3.8.1</version>

     <scope>test</scope>

   </dependency>



   <dependency>

     <groupId>org.apache.spark</groupId>

     <artifactId>spark-core_2.11</artifactId>

     <version>2.0.0</version>

   </dependency>



   <dependency>

     <groupId>org.mongodb.spark</groupId>

     <artifactId>mongo-spark-connector_2.11</artifactId>

     <version>2.2.3</version>

   </dependency>



    <dependency>

        <groupId>org.apache.spark</groupId>

        <artifactId>spark-sql_2.11</artifactId>

        <version>2.0.0</version>

    </dependency>



    <dependency>

        <groupId>com.google.code.gson</groupId>

        <artifactId>gson</artifactId>

        <version>2.8.5</version>

    </dependency>





    <dependency>

        <groupId>org.mongodb</groupId>

        <artifactId>bson</artifactId>

        <version>3.8.0</version>

    </dependency>





    <dependency>

        <groupId>org.apache.spark</groupId>

        <artifactId>spark-streaming_2.11</artifactId>

        <version>2.0.0</version>

    </dependency>

    <dependency>

        <groupId>org.apache.spark</groupId>

        <artifactId>spark-sql-kafka-0-10_2.11</artifactId>

        <version>2.0.2</version>

    </dependency>





    <dependency>

        <groupId>org.apache.spark</groupId>

        <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>

        <version>2.0.0</version>

    </dependency>



    <!-- <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka_2.11</artifactId> 

        <version>1.6.3</version> </dependency> -->



    <dependency>

        <groupId>org.apache.kafka</groupId>

        <artifactId>kafka-clients</artifactId>

        <version>1.1.0</version>

    </dependency>









    <dependency>

        <groupId>com.fasterxml.jackson.module</groupId>

        <artifactId>jackson-module-scala_2.11</artifactId>

        <version>2.9.5</version>

    </dependency>



</dependencies>

How do I fix this? Please help me out.