title: Spark. ~100 million rows. Size exceeds Integer.MAX_VALUE?
body: <p>(This is with Spark 2.0 running on a small three machine Amazon EMR cluster)</p>
  
  <p>I have a PySpark job that loads some large text files into a Spark RDD, does count() which successfully returns 158,598,155.</p>
  
  <p>Then the job parses each row into a pyspark.sql.Row instance, builds a DataFrame, and does another count. This second count() on the DataFrame causes an exception in Spark internal code <code>Size exceeds Integer.MAX_VALUE</code>. This works with smaller volumes of data. Can someone explain why/how this would happen?</p>
  
  <pre><code>org.apache.spark.SparkException: Job aborted due to stage failure: Task 22 in stage 1.0 failed 4 times, most recent failure: Lost task 22.3 in stage 1.0 (TID 77, ip-172-31-97-24.us-west-2.compute.internal): java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE
      at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:869)
      at org.apache.spark.storage.DiskStore$$anonfun$getBytes$2.apply(DiskStore.scala:103)
      at org.apache.spark.storage.DiskStore$$anonfun$getBytes$2.apply(DiskStore.scala:91)
      at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1287)
      at org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:105)
      at org.apache.spark.storage.BlockManager.getLocalValues(BlockManager.scala:439)
      at org.apache.spark.storage.BlockManager.get(BlockManager.scala:604)
      at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:661)
      at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:330)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:281)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47)
      at org.apache.spark.scheduler.Task.run(Task.scala:85)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  </code></pre>
  
  <p>PySpark code:</p>
  
  <pre><code>raw_rdd = spark_context.textFile(full_source_path)
  
  # DEBUG: This call to count() is expensive
  # This count succeeds and returns 158,598,155
  logger.info("raw_rdd count = %d", raw_rdd.count())
  logger.info("completed getting raw_rdd count!!!!!!!")
  
  row_rdd = raw_rdd.map(row_parse_function).filter(bool)
  data_frame = spark_sql_context.createDataFrame(row_rdd, MySchemaStructType)
  
  data_frame.cache()
  # This will trigger the Spark internal error
  logger.info("row count = %d", data_frame.count())
  </code></pre>