Why does format(“kafka”) fail with “Failed to find data source: kafka.” (even with uber-jar)?

by

kafka data source is an external module and is not available to Spark applications by default.

You have to define it as a dependency in your pom.xml (as you have done), but that’s just the very first step to have it in your Spark application.

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
        <version>2.2.0</version>
    </dependency>

With that dependency you have to decide whether you want to create a so-called uber-jar that would have all the dependencies bundled altogether (that results in a fairly big jar file and makes the submission time longer) or use --packages (or less flexible --jars) option to add the dependency at spark-submit time.

(There are other options like storing the required jars on Hadoop HDFS or using Hadoop distribution-specific ways of defining dependencies for Spark applications, but let’s keep things simple)

I’d recommend using --packages first and only when it works consider the other options.

Use spark-submit --packages to include the spark-sql-kafka-0-10 module as follows.

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0

Include the other command-line options as you wish.

Uber-Jar Approach

Including all the dependencies in a so-called uber-jar may not always work due to how META-INF directories are handled.

For kafka data source to work (and other data sources in general) you have to ensure that META-INF/services/org.apache.spark.sql.sources.DataSourceRegister of all the data sources are merged (not replace or first or whatever strategy you use).

kafka data sources uses its own META-INF/services/org.apache.spark.sql.sources.DataSourceRegister that registers org.apache.spark.sql.kafka010.KafkaSourceProvider as the data source provider for kafka format.

More Related Contents:

  1. How to get Kafka offsets for structured query for manual and reliable offset management?
  2. How to melt Spark DataFrame?
  3. Spark SQL: apply aggregate functions to a list of columns
  4. How to connect Spark SQL to remote Hive metastore (via thrift protocol) with no hive-site.xml?
  5. Unpivot in spark-sql/pyspark
  6. Spark SQL replacement for MySQL’s GROUP_CONCAT aggregate function
  7. Spark SQL – load data with JDBC using SQL statement, not table name
  8. Spark Dataframe validating column names for parquet writes
  9. How to optimize partitioning when migrating data from JDBC source?
  10. How to check if spark dataframe is empty?
  11. How to read records in JSON format from Kafka using Structured Streaming?
  12. How to connect to remote hive server from spark [duplicate]
  13. Filtering a spark dataframe based on date
  14. Pyspark : forward fill with last observation for a DataFrame
  15. Spark DataFrame Schema Nullable Fields
  16. PySpark: how to resample frequencies
  17. multiple conditions for filter in spark data frames
  18. Why does Spark think this is a cross / Cartesian join
  19. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  20. Spark SQL broadcast hash join
  21. Rename more than one column using withColumnRenamed
  22. What is the difference between Apache Spark SQLContext vs HiveContext?
  23. What should be the optimal value for spark.sql.shuffle.partitions or how do we increase partitions when using Spark SQL?
  24. Spark lists all leaf node even in partitioned data
  25. Array Intersection in Spark SQL
  26. PySpark: How to fillna values in dataframe for specific columns?
  27. Why spark application fail with “executor.CoarseGrainedExecutorBackend: Driver Disassociated”?
  28. What does “Correlated scalar subqueries must be Aggregated” mean?
  29. How to calculate Median in spark sqlContext for column of data type double
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment