title: scala.MatchError: [abc,cde,null,3] (of class org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema) in Spark JSON with missing fields
body: <p>I have as JSON input file:</p>
  
  <pre><code>{"a": "abc", "b": "bcd", "d": 3},
  {"a": "ezx", "b": "hdg", "c": "ssa"},
  ...
  </code></pre>
  
  <p>some fields per object are missing instead of placing the <code>null</code> value.</p>
  
  <p>In Apache Spark using Scala:</p>
  
  <pre><code>import SparkCommons.sparkSession.implicits._
  
  private val inputJsonPath: String = "resources/input/input.json"
  
  private val schema = StructType(Array(
    StructField("a", StringType, nullable = false),
    StructField("b", StringType, nullable = false),
    StructField("c", StringType, nullable = true),
    StructField("d", DoubleType, nullable = true)
  ))
  
  private val inputDF: DataFrame = SparkCommons.sparkSession
    .read
    .schema(schema)
    .json(inputJsonPath)
    .cache()
  
  inputDF.printSchema()
  
  val dataRdd = inputDF.rdd
  .map {
    case Row(a: String, b: String, c: String, d: Double) =&gt;
      MyCaseClass(a, b, c, d)
  }
  
  val dataMap = dataRdd.collectAsMap()
  </code></pre>
  
  <p>The <code>MyCaseClass</code> code:</p>
  
  <pre><code>case class MyCaseClass(
                a: String,
                b: String,
                c: String = null,
                d: Double = Predef.Double2double(null)
  )
  </code></pre>
  
  <p>I get the following schema as output:</p>
  
  <pre><code>root
   |-- a: string (nullable = true)
   |-- b: string (nullable = true)
   |-- c: string (nullable = true)
   |-- d: double (nullable = true)
  </code></pre>
  
  <p>Program compiles but at runtime once Spark is doing the jobs I get the following exception:</p>
  
  <pre><code>[error] - org.apache.spark.executor.Executor - Exception in task 3.0 in stage 4.0 (TID 21)
  scala.MatchError: [abc,bcd,null,3] (of class org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema)
  at com.matteoguarnerio.spark.SparkOperations$$anonfun$1.apply(SparkOperations.scala:62) ~[classes/:na]
  at com.matteoguarnerio.spark.SparkOperations$$anonfun$1.apply(SparkOperations.scala:62) ~[classes/:na]
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:410) ~[scala-library-2.11.11.jar:na]
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:410) ~[scala-library-2.11.11.jar:na]
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:410) ~[scala-library-2.11.11.jar:na]
  at org.apache.spark.util.random.SamplingUtils$.reservoirSampleAndCount(SamplingUtils.scala:42) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.RangePartitioner$$anonfun$9.apply(Partitioner.scala:261) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.RangePartitioner$$anonfun$9.apply(Partitioner.scala:259) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$25.apply(RDD.scala:820) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$25.apply(RDD.scala:820) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.rdd.RDD.iterator(RDD.scala:283) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.scheduler.Task.run(Task.scala:86) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274) ~[spark-core_2.11-2.0.2.jar:2.0.2]
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_144]
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_144]
  at java.lang.Thread.run(Thread.java:748) [na:1.8.0_144]
  </code></pre>
  
  <p>Spark version: 2.0.2</p>
  
  <p>Scala version: 2.11.11</p>
  
  <ul>
  <li>How to solve this exception and iterate over even if some fields are <code>null</code> or missing in the RDD matching and creating objects?</li>
  <li>Why the schema, even if I explicitly defined not nullable and nullable on some fields, is everything nullable?</li>
  </ul>
  
  <p><strong>UPDATE</strong></p>
  
  <p>I just used a workaround on <code>dataRdd</code> to avoid the problem:</p>
  
  <pre><code>private val dataRdd = inputDF.rdd
  .map {
    case r: GenericRowWithSchema =&gt; {
        val a = r.getAs("a").asInstanceOf[String]
        val b = r.getAs("b").asInstanceOf[String]
  
        var c: Option[String] = None
        var d: Option[Double] = None
  
        try {
          c = if (r.isNullAt(r.fieldIndex("c"))) None: Option[String] else Some(r.getAs("c").asInstanceOf[String])
          d = if (r.isNullAt(r.fieldIndex("d"))) None: Option[Double] else Some(r.getAs("d").asInstanceOf[Double])
        } catch {
          case _: Throwable =&gt; None
        }
  
        MyCaseClass(a, b, c, d)
    }
  }
  </code></pre>
  
  <p>and changed <code>MyCaseClass</code> in this way:</p>
  
  <pre><code>case class MyCaseClass(
                a: String,
                b: String,
                c: Option[String],
                d: Option[Double]
  )
  </code></pre>