Why is spark throwing an ArrayIndexOutOfBoundsException expection for empty attributes?

Question

Context

I am using Spark 1.5.

I have a file records.txt which is ctrl A delimited and in that file 31st index is for subscriber_id. For some records the subscriber_id is empty. Record with subscriber_id is NOT empty.

Here subscriber_id(UK8jikahasjp23) is located at one before the last attribute:

99^A2013-12-11^A23421421412^qweqweqw2222^A34232432432^A365633049^A1^A6yudgfdhaf9923^AAC^APrimary DTV^AKKKR DATA+ PVR3^AGrundig^AKKKR PVR3^AKKKR DATA+ PVR3^A127b146^APVR3^AYes^ANo^ANo^ANo^AYes^AYes^ANo^A2017-08-07 21:27:30.000000^AYes^ANo^ANo^A6yudgfdhaf9923^A7290921396551747605^A2013-12-11 16:00:03.000000^A7022497306379992936^AUK8jikahasjp23^A

Record with subscriber_id is empty:

23^A2013-12-11^A23421421412^qweqweqw2222^A34232432432^A365633049^A1^A6yudgfdhaf9923^AAC^APrimary DTV^AKKKR DATA+ PVR3^AGrundig^AKKKR PVR3^AKKKR DATA+ PVR3^A127b146^APVR3^AYes^ANo^ANo^ANo^AYes^AYes^ANo^A2017-08-07 21:27:30.000000^AYes^ANo^ANo^A6yudgfdhaf9923^A7290921396551747605^A2013-12-11 16:00:03.000000^A7022497306379992936^A^A

Problem

I am getting java.lang.ArrayIndexOutOfBoundsException for the records with empty subscriber_id.

Why is spark throwing java.lang.ArrayIndexOutOfBoundsException for the empty values for the field subscriber_id?

16/08/20 10:22:18 WARN scheduler.TaskSetManager: Lost task 31.0 in stage 8.0 : java.lang.ArrayIndexOutOfBoundsException: 31

 case class CustomerCard(accountNumber:String, subscriber_id:String,subscriptionStatus:String )

     object CustomerCardProcess {
    val log = LoggerFactory.getLogger(this.getClass.getName)


   def doPerform(sc: SparkContext, sqlContext: HiveContext, custCardRDD: RDD[String]): DataFrame = {

    import sqlContext.implicits._
    log.info("doCustomerCardProcess method started")
     val splitRDD        =    custCardRDD.map(elem => elem.split("\\u0001"))
     val schemaRDD       =    splitRDD.map(arr => new CustomerCard( arr(3).trim, arr(31).trim,arr(8).trim))

schemaRDD.toDF().registerTempTable("customer_card")
val custCardDF = sqlContext.sql(
  """
    |SELECT
    |accountNumber,
    |subscriber_id
    |FROM
    |customer_card
    |WHERE
    |subscriptionStatus IN('AB', 'AC', 'PC')
    |AND accountNumber IS NOT NULL AND LENGTH(accountNumber) > 0
  """.stripMargin)

log.info("doCustomerCardProcess method ended")
custCardDF
  }

}

Error

13/09/12 23:22:18 WARN scheduler.TaskSetManager: Lost task 31.0 in stage 8.0 (TID 595, : java.lang.ArrayIndexOutOfBoundsException: 31 at com.org.CustomerCardProcess$$anonfun$2.apply(CustomerCardProcess.scala:23) at com.org.CustomerCardProcess$$anonfun$2.apply(CustomerCardProcess.scala:23) at scala.collection.Iterator$$anon$11.next(Iterator.scala:328) at scala.collection.Iterator$$anon$11.next(Iterator.scala:328) at scala.collection.Iterator$$anon$14.hasNext(Iterator.scala:389) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.insertAll(BypassMergeSortShuffleWriter.java:118) at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:73) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41) at org.apache.spark.scheduler.Task.run(Task.scala:88) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745)

Could anyone help me to fix this issue ?

Ramesh Maharjan Ramesh Maharjan · Accepted Answer · 2017-08-30T14:10:35

The split function is neglecting all the empty fields at the end of splitted line. So,

Change your following line

 val splitRDD = custCardRDD.map(elem => elem.split("\\u0001"))

to

val splitRDD = custCardRDD.map(elem => elem.split("\\u0001", -1))

-1 tells to consider all the empty fields.

Why is spark throwing an ArrayIndexOutOfBoundsException expection for empty attributes?

1 Answers