CODEWITHSUNDEEP
javaapache-sparkhadoop
Ahmad Alhilal
Ahmad Alhilal

Reputation: 454

Integration of spark and kafka, exception in Spark-submit a jar

Got a NullPointerException when submit a jar file that integrate kafka with spark under ubuntu. I'm trying to run the code at https://github.com/apache/spark/tree/v2.1.1/examples

I tried to check if installing spark under Ubuntu requires set HADOOP_HOME; However, HADOOP_HOME is not set, checked carefully the arguments of the jar. 

./bin/spark-submit --class "org.apache.spark.examples.streaming.JavaKafkaWordCount" --packages org.apache.spark:spark-streaming-kafka-0-10_2.11:2.1.0 --master local[*] --jars ~/software/JavaKafkaWordCount.jar localhost:2181 test-consumer-group streams-plaintext-input 1

Exception in thread "main" java.lang.NullPointerException at org.apache.hadoop.fs.Path.getName(Path.java:337) at org.apache.spark.deploy.DependencyUtils$.downloadFile(DependencyUtils.scala:136) at org.apache.spark.deploy.SparkSubmit$$anonfun$prepareSubmitEnvironment$7.apply(SparkSubmit.scala:367) at org.apache.spark.deploy.SparkSubmit$$anonfun$prepareSubmitEnvironment$7.apply(SparkSubmit.scala:367) at scala.Option.map(Option.scala:146) at org.apache.spark.deploy.SparkSubmit.prepareSubmitEnvironment(SparkSubmit.scala:366) at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:143) at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86) at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:924) at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:933) at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

Upvotes: 1

Views: 1384

Answers (1)

Ram Ghadiyaram
Ram Ghadiyaram

Reputation: 29165

your path uri jar is not understandable see this DependencyUtils.scala#L136

 /**
   * Download a file from the remote to a local temporary directory. If the input path points to
   * a local path, returns it with no operation.
   *
   * @param path A file path from where the files will be downloaded.
   * @param targetDir A temporary directory for which downloaded files.
   * @param sparkConf Spark configuration.
   * @param hadoopConf Hadoop configuration.
   * @param secMgr Spark security manager.
   * @return Path to the local file.
   */
  def downloadFile(
      path: String,
      targetDir: File,
      sparkConf: SparkConf,
      hadoopConf: Configuration,
      secMgr: SecurityManager): String = {
    require(path != null, "path cannot be null.")
    val uri = Utils.resolveURI(path)

    uri.getScheme match {
      case "file" | "local" => path
      case "http" | "https" | "ftp" if Utils.isTesting =>
        // This is only used for SparkSubmitSuite unit test. Instead of downloading file remotely,
        // return a dummy local path instead.
        val file = new File(uri.getPath)
        new File(targetDir, file.getName).toURI.toString
      case _ =>
        val fname = new Path(uri).getName()
        val localFile = Utils.doFetchFile(uri.toString(), targetDir, fname, sparkConf, secMgr,
          hadoopConf)
        localFile.toURI().toString()
    }
  }

In your spark-submit change the args like this

--jars /fullpath/JavaKafkaWordCount.jar instead of --jars ~/software/JavaKafkaWordCount.jar

Upvotes: 2

Related Questions