Unable to query MongoDB using Spark SQL via MongoDB Connector

Question

The eventual goal that I want to achieve is that I want to query my MongoDB Collection through Spark SQL using Scala Code as an Independent application. I have successfully installed Spark on my local which is running "Windows 10" operating system. I can run spark-shell, Spark Master node and worker node. So from the looks of it, the apache spark is working fine on my p.c

I can also query my MongoDB collection by running the scala code in Spark Shell.

Problem:

When I try to use the same code from my Scala project using MongoDB Spark Connector for scala. I am running into an error which I am unable to figure out. IT seems like an environment issue, I looked it up and many people suggested that it happens if you use Java 9 or higher version. I am using Java 8 so that's not the issue in my case. That is why I have also posted my java -version snapshot in the post.

But when I run the code, I get the following error. It would be great help IF somebody can advise me in any direction.

Scala Code:

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import com.mongodb.spark.config._
import com.mongodb.spark._

object SparkSQLMongoDBConnector {

  def main(args: Array[String]): Unit ={

    var sc: SparkContext = null

    var conf = new SparkConf()
    conf.setAppName("MongoSparkConnectorIntro")
      .setMaster("local")
      .set("spark.hadoop.validateOutputSpecs", "false")
        .set("spark.mongodb.input.uri","mongodb://127.0.0.1/metadatastore.metadata_collection?readPreference=primaryPreferred")
      .set("spark.mongodb.output.uri","mongodb://127.0.0.1/metadatastore.metadata_collection?readPreference=primaryPreferred")

    sc = new SparkContext(conf)
    val spark = SparkSession.builder().master("spark://192.168.137.221:7077").appName("MongoSparkConnectorIntro").config("spark.mongodb.input.uri", "mongodb://127.0.0.1/metadatastore.metadata_collection?readPreference=primaryPreferred").config("spark.mongodb.output.uri", "mongodb://127.0.0.1/metadatastore.metadata_collection?readPreference=primaryPreferred").getOrCreate()


    val readConfig = ReadConfig(Map("collection" -> "spark", "readPreference.name" -> "secondaryPreferred"), Some(ReadConfig(sc)))
    val customRdd = MongoSpark.load(sc, readConfig)

    println(customRdd.count)
    println(customRdd.first.toString())


  }

}

SBT:

 scalaVersion := "2.12.8"
    libraryDependencies += "org.mongodb.spark" %% "mongo-spark-connector" % "2.4.0"
    libraryDependencies += "org.apache.spark" %% "spark-core" % "2.4.0"
    libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.4.0"

Java Version:

Error:

This is the error that I face when I run the Scala code in the IntelliJ.

Exception in thread "main" java.lang.ExceptionInInitializerError
    at org.apache.hadoop.util.StringUtils.<clinit>(StringUtils.java:79)
    at org.apache.hadoop.security.Groups.parseStaticMapping(Groups.java:116)
    at org.apache.hadoop.security.Groups.<init>(Groups.java:93)
    at org.apache.hadoop.security.Groups.<init>(Groups.java:73)
    at org.apache.hadoop.security.Groups.getUserToGroupsMappingService(Groups.java:293)
    at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:283)
    at org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:260)
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInorg.apache.spark.util.Utils$.$anonfun$getCurrentUserName$1(Utils.scala:2422)
    at scala.Option.getOrElse(Option.scala:138)
    at org.apache.spark.util.Utils$.getCurrentUserName(Utils.scala:2422)
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:293)
    at SparkSQLMongoDBConnector$.main(SparkSQLMongoDBConnector.scala:35)
    at SparkSQLMongoDBConnector.main(SparkSQLMongoDBConnector.scala)
Caused by: java.lang.StringIndexOutOfBoundsException: begin 0, end 3, length 2
    at java.base/java.lang.String.checkBoundsBeginEnd(String.java:3410)
    at java.base/java.lang.String.substring(String.java:1883)
    at org.apache.hadoop.util.Shell.<clinit>(Shell.java:50)
    ... 16 more

Any help would be much appreciated.

Answer 1

Shell checks your Java version via java.version variable

private static boolean IS_JAVA7_OR_ABOVE =
  System.getProperty("java.version").substring(0, 3).compareTo("1.7") >= 0;

Make sure it is defined.

This line was changed in Hadoop 2.7+, but by default, Spark uses 2.6.5.