title: Why does Spark application fail with “ClassNotFoundException: Failed to find data source: kafka” as uber-jar with sbt assembly?
body: <p>I'm trying to run a sample like <a href="https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/sql/streaming/StructuredKafkaWordCount.scala" rel="nofollow noreferrer">StructuredKafkaWordCount</a>. I started with the <a href="http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html" rel="nofollow noreferrer">Spark Structured Streaming Programming guide</a>. </p>
  
  <p>My code is</p>
  
  <pre><code>package io.boontadata.spark.job1
  
  import org.apache.spark.sql.SparkSession
  
  object DirectKafkaAggregateEvents {
    val FIELD_MESSAGE_ID = 0
    val FIELD_DEVICE_ID = 1
    val FIELD_TIMESTAMP = 2
    val FIELD_CATEGORY = 3
    val FIELD_MEASURE1 = 4
    val FIELD_MEASURE2 = 5
  
    def main(args: Array[String]) {
      if (args.length &lt; 3) {
        System.err.println(s"""
          |Usage: DirectKafkaAggregateEvents &lt;brokers&gt; &lt;subscribeType&gt; &lt;topics&gt;
          |  &lt;brokers&gt; is a list of one or more Kafka brokers
          |  &lt;subscribeType&gt; sample value: subscribe
          |  &lt;topics&gt; is a list of one or more kafka topics to consume from
          |
          """.stripMargin)
        System.exit(1)
      }
  
      val Array(bootstrapServers, subscribeType, topics) = args
  
      val spark = SparkSession
        .builder
        .appName("boontadata-spark-job1")
        .getOrCreate()
  
      import spark.implicits._
  
      // Create DataSet representing the stream of input lines from kafka
      val lines = spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", bootstrapServers)
        .option(subscribeType, topics)
        .load()
        .selectExpr("CAST(value AS STRING)")
        .as[String]
  
      // Generate running word count
      val wordCounts = lines.flatMap(_.split(" ")).groupBy("value").count()
  
      // Start running the query that prints the running counts to the console
      val query = wordCounts.writeStream
        .outputMode("complete")
        .format("console")
        .start()
  
      query.awaitTermination()
    }
  
  }
  </code></pre>
  
  <p>I added the following sbt files: </p>
  
  <p>build.sbt: </p>
  
  <pre><code>name := "boontadata-spark-job1"
  version := "0.1"
  scalaVersion := "2.11.7"
  
  libraryDependencies += "org.apache.spark" % "spark-core_2.11" % "2.0.2" % "provided"
  libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % "2.0.2" % "provided"
  libraryDependencies += "org.apache.spark" % "spark-sql_2.11" % "2.0.2" % "provided"
  libraryDependencies += "org.apache.spark" % "spark-sql-kafka-0-10_2.11" % "2.0.2"
  libraryDependencies += "org.apache.spark" % "spark-streaming-kafka-0-10_2.11" % "2.0.2"
  libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.10.1.1"
  libraryDependencies += "org.apache.kafka" % "kafka_2.11" % "0.10.1.1"
  
  // META-INF discarding
  assemblyMergeStrategy in assembly := { 
     {
      case PathList("META-INF", xs @ _*) =&gt; MergeStrategy.discard
      case x =&gt; MergeStrategy.first
     }
  }
  </code></pre>
  
  <p>I also added project/assembly.sbt</p>
  
  <pre><code>addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.3")
  </code></pre>
  
  <p>This creates a Uber jar with the non <code>provided</code> jars. </p>
  
  <p>I submit with the following line: </p>
  
  <pre><code>spark-submit boontadata-spark-job1-assembly-0.1.jar ks1:9092,ks2:9092,ks3:9092 subscribe sampletopic
  </code></pre>
  
  <p>but I get this runtime error: </p>
  
  <pre><code>Exception in thread "main" java.lang.ClassNotFoundException: Failed to find data source: kafka. Please find packages at https://cwiki.apache.org/confluence/display/SPARK/Third+Party+Projects
          at org.apache.spark.sql.execution.datasources.DataSource.lookupDataSource(DataSource.scala:148)
          at org.apache.spark.sql.execution.datasources.DataSource.providingClass$lzycompute(DataSource.scala:79)
          at org.apache.spark.sql.execution.datasources.DataSource.providingClass(DataSource.scala:79)
          at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:218)
          at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:80)
          at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:80)
          at org.apache.spark.sql.execution.streaming.StreamingRelation$.apply(StreamingRelation.scala:30)
          at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:124)
          at io.boontadata.spark.job1.DirectKafkaAggregateEvents$.main(StreamingJob.scala:41)
          at io.boontadata.spark.job1.DirectKafkaAggregateEvents.main(StreamingJob.scala)
          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.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:736)
          at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:185)
          at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:210)
          at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:124)
          at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
  Caused by: java.lang.ClassNotFoundException: kafka.DefaultSource
          at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
          at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
          at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
          at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$5$$anonfun$apply$1.apply(DataSource.scala:132)
          at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$5$$anonfun$apply$1.apply(DataSource.scala:132)
          at scala.util.Try$.apply(Try.scala:192)
          at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$5.apply(DataSource.scala:132)
          at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$5.apply(DataSource.scala:132)
          at scala.util.Try.orElse(Try.scala:84)
          at org.apache.spark.sql.execution.datasources.DataSource.lookupDataSource(DataSource.scala:132)
          ... 18 more
  16/12/23 13:32:48 INFO spark.SparkContext: Invoking stop() from shutdown hook
  </code></pre>
  
  <p>Is there a way to know which class is not found so that I can search the maven.org repo for that class. </p>
  
  <p>The <code>lookupDataSource</code> source code seems to be at line 543 at <a href="https://github.com/apache/spark/blob/83a6ace0d1be44f70e768348ae6688798c84343e/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala" rel="nofollow noreferrer">https://github.com/apache/spark/blob/83a6ace0d1be44f70e768348ae6688798c84343e/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala</a> but I couldn't find a direct link with Kafka data source...</p>
  
  <p>Complete source code is here: <a href="https://github.com/boontadata/boontadata-streams/tree/ad0d0134ddb7664d359c8dca40f1d16ddd94053f" rel="nofollow noreferrer">https://github.com/boontadata/boontadata-streams/tree/ad0d0134ddb7664d359c8dca40f1d16ddd94053f</a></p>