[FIXED] PySpark / Kafka - org.apache.spark.sql.kafka010.KafkaSourceProvider could not be instantiated

Issue

So, I'm working on setting up a development environment for working with PySpark and Kafka. I'm working through getting things setup so I can run these tutorials in a Jupyter notebook as a 'hello world' exercise: https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html

Unfortunately, I'm currently hitting the following error when I attempt to connect to the Kafka stream:

Py4JJavaError: An error occurred while calling o68.load.
: java.util.ServiceConfigurationError: org.apache.spark.sql.sources.DataSourceRegister: Provider org.apache.spark.sql.kafka010.KafkaSourceProvider could not be instantiated
    at java.base/java.util.ServiceLoader.fail(ServiceLoader.java:583)
    at java.base/java.util.ServiceLoader$ProviderImpl.newInstance(ServiceLoader.java:805)
    at java.base/java.util.ServiceLoader$ProviderImpl.get(ServiceLoader.java:723)
    at java.base/java.util.ServiceLoader$3.next(ServiceLoader.java:1395)
    ...

Now, some digging has told me that the most common cause of this issue is version mismatches (either for the Spark, or Scala versions in use). However, I'm able to confirm that these are aligned properly:

Spark: 3.1.1
Scala: 2.12.10

conf/spark-defaults.conf

...
spark.jars.packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1

EDIT So, some additional observations from trying to figure this out:

• It looks like this is at least partially a Jupyter notebook issue, as I can now get things working just fine via the pyspark shell.
• Looks like my notebook is firing up its own instance of Spark, so maybe there's some difference in how Spark is being run there vs from a terminal window?
• At a loss for how they're different though, as both environments should be using mostly default configurations.
• Ok - looks like it doesn't work when invoked via the regular Python REPL either, which is leading me to think there's something different about the spark context being created by the pyspark shell and the one I'm creating in my notebook.
• Ok - looks like something differs when things are run via Jupyter - hadoop.common.configuration.version has a value of 0.23.0 for the notebook instance, but 3.0.0 for the pyspark shell instance. Not sure why this might be or what it may mean yet.

What else should I check to confirm that this is setup correctly?

Solution

Ok - so it looks like the difference was that findspark was locating and using a different Spark Home directory (one that came installed with the pyspark installation via pip).

It also looks like Spark 3.1.1 for Hadoop 2.7 has issues with the Kafka client (or maybe needs to be configured differently) but Spark 3.1.1 for Hadoop 3.2 works fine.

Solution was to ensure that I explicitly chose my SPARK_HOME by passing the spark_home path to findspark.init()

findspark.init(spark_home='/path/to/desired/home')

Things to watch out for that got me and might trip you up too:

• If you've installed pyspark through pip / mambaforge this will also deploy a second SPARK_HOME - this can create dependency / library confusion.
• Many of the scripts in bin/ use SPARK_HOME to determine where to execute, so don't assume that just because you're running a script from one home that you're running spark IN that home.

Answered By - Adam Luchjenbroers