Why does executing Structured Streaming application fail with "Failed to find data source: kafka"?

Question

I am trying to connect Spark Structured Streaming with kafka and it throws the below error:

Exception in thread "main" java.lang.ClassNotFoundException: Failed to find data source: kafka. Please find packages at ...

Based on the documentation I have added the required dependencies

and my kafka and zookeeper servers are running. Not sure what the issue is. Also, I am using it this way

import spark.implicits._
val feedback =spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:2181").option("subscribe", "kafka_input_topic")
      .load().as[InputMessage].filter(_.lang.equals("en"))

Any help is appreciated. Thank you

Do you have `"org.apache.spark" % "spark-sql-kafka-0-10_2.11" % SPARK_VERSION` added to your dependency list? (SPARK_VERSION is a placeholder for your proper spark version). — Yuval Itzchakov, Feb 11 '18 at 11:52
yes. @YuvalItzchakov ` org.apache.spark spark-sql-kafka-0-10_2.11 2.2.0 provided ` — Rahul Kumar, Feb 11 '18 at 11:55

score 4 · Answer 1 · answered Feb 11 '18 at 11:56

4

The problem, as you mentioned in your comments, is this:

<scope>provided</scope>

Remove the provided scope for sql-kafka, as it is not provided by the Spark installation.

answered Feb 11 '18 at 11:56

Yuval Itzchakov

146,575
32
257
321

still the same error – Rahul Kumar Feb 11 '18 at 12:04
Did you refresh your dependencies properly? Are you compiling an uber jar? – Yuval Itzchakov Feb 11 '18 at 12:06
yes. its an uber jar.. but the error doesnt show up when im adding this to spark submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 – Rahul Kumar Feb 11 '18 at 12:18
@RahulKumar That means something in your packaging didn't refresh properly, and it still caches the version with `provided`. – Yuval Itzchakov Feb 11 '18 at 12:19
@RahulKumar check this thread https://stackoverflow.com/questions/48011941/why-does-formatkafka-fail-with-failed-to-find-data-source-kafka-with-ube – Kleyson Rios Feb 12 '18 at 09:10

score 1 · Answer 2 · answered Jun 08 '18 at 13:08

1

you could use the kafka data source by the fully-qualified name (not the alias) as follows:

spark.readStream.format("org.apache.spark.sql.kafka010.KafkaSourceProvider").load

answered Jun 08 '18 at 13:08

Punk Monday

11
1

This is not work for me, I was really hoping for an option like this – Aaron Stainback Jun 30 '20 at 00:47

score 1 · Answer 3 · edited Jun 20 '20 at 09:12

The issue is that the necessary jar is not included in CLASSPATH at runtime (not build time).

Based on the documentation you linked to you added the required dependencies to your build definition file (pom.xml or build.sbt or build.gradle), but the exception happens while you try to run the application which is after it is built, doesn't it?

What you miss is that part of the documentation about deployment, i.e. Deploying:

As with any Spark applications, spark-submit is used to launch your application. spark-sql-kafka-0-10_2.11 and its dependencies can be directly added to spark-submit using --packages, such as,
./bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 ..

You have to add this --packages or you'd have to create an uber-jar that would make the dependency part of your jar file.

score 0 · Answer 4 · answered Mar 12 '18 at 12:30

If using maven then the following way of building jar with dependencies might solve your issue.

Add the spark dependencies like below:

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.2.1</version>
        <scope>${spark.scope}</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
        <version>2.2.1</version>
    </dependency>

Then configure your maven profiles as below:

<profiles>
    <profile>
        <id>default</id>
        <properties>
            <profile.id>dev</profile.id>
            <spark.scope>compile</spark.scope>
        </properties>
        <activation>
            <activeByDefault>true</activeByDefault>
        </activation>
    </profile>
    <profile>
        <id>test</id>
        <properties>
            <profile.id>test</profile.id>
            <spark.scope>provided</spark.scope>
        </properties>
    </profile>
    <profile>
        <id>online</id>
        <properties>
            <profile.id>online</profile.id>
            <spark.scope>provided</spark.scope>
        </properties>
    </profile>
</profiles>

Add the followign plugin:

<plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>3.1.0</version>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id> <!-- this is used for inheritance merges -->
                    <phase>package</phase> <!-- bind to the packaging phase -->
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>

Then buld your jar using mvn clean install -Ponline -DskipTests. This should solve your issue

Why does executing Structured Streaming application fail with "Failed to find data source: kafka"?

4 Answers4