title: Handling corrupt JSON rows in Spark 2.11 - different behaviour than 1.6
body: <p>We have snappy files that we read with sql context. e.g.</p>
  
  <pre><code>val sqlContext = new org.apache.spark.sql.SQLContext(sc)
  val df = sqlContext.read.json("s3://bucket/problemfile.snappy")
  </code></pre>
  
  <p>In spark 1.6 we would handle corrupt records by something like the below:</p>
  
  <pre><code>invalidJSON = rawEvents.select("*").where("_corrupt_record is not null");
  validJSON = rawEvents.select("*").where("_corrupt_record is null");
  </code></pre>
  
  <p>In Spark 2.11, we are not even able to read the corrupted record e.g</p>
  
  <pre><code>scala&gt; df.select("*").where("_corrupt_record is null").count()
  
  18/03/31 00:45:06 ERROR TaskSetManager: Task 0 in stage 1.0 failed 4 times; aborting job 
  org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1.0 (TID 4, ip-172-31-48-73.ec2.internal, executor 2): 
  java.io.CharConversionException: Unsupported UCS-4 endianness (3412) detected
  at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.reportWeirdUCS4(ByteSourceJsonBootstrapper.java:469)
  at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.checkUTF32(ByteSourceJsonBootstrapper.java:434)
  at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.detectEncoding(ByteSourceJsonBootstrapper.java:141)
  at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.constructParser(ByteSourceJsonBootstrapper.java:215)
  at com.fasterxml.jackson.core.JsonFactory._createParser(JsonFactory.java:1287)
  </code></pre>
  
  <p>I know we can set spark.sql.files.ignoreCorruptFiles=true in 2.X but that we'd potentially lose records depending on where the corrupted record was. </p>
  
  <p>Is there any other way we can skip over the corrupted record? </p>
  
  <p>Thanks</p>