title: 'Premature end of Content-Length' with Spark Application using s3a
body: <p>I'm writing a Spark based application which works around a pretty huge data stored on s3. It's about <strong>15 TB</strong> in size uncompressed. Data is laid across multiple small LZO compressed files files, varying from 10-100MB.</p>
  
  <p>By default the job spawns 130k tasks while reading dataset and mapping it to schema.</p>
  
  <p>And then it fails around 70k tasks completions and after ~20 tasks failure.</p>
  
  <p><strong>Exception:</strong></p>
  
  <pre><code>WARN lzo.LzopInputStream: IOException in getCompressedData; likely LZO corruption.
      org.apache.http.ConnectionClosedException: Premature end of Content-Length delimited message body
  </code></pre>
  
  <p>Looks like the s3 connection is getting closed prematurely.</p>
  
  <p><strong>I have tried nearly 40 different combos of configurations.</strong></p>
  
  <p><strong>To summarize them:</strong> 1 executor to 3 executors per node, 18GB to 42GB <code>--executor-memory</code>, 3-5 <code>--executor-cores</code>, 1.8GB-4.0 GB <code>spark.yarn.executor.memoryOverhead</code>, Both, Kryo and Default Java serializers, 0.5 to 0.35 <code>spark.memory.storageFraction</code>, default, 130000 to 200000 partitions for bigger dataset. default, 200 to 2001 <code>spark.sql.shuffle.partitions</code>.</p>
  
  <p><strong>And most importantly:</strong> 100 to 2048 <code>fs.s3a.connection.maximum</code> property.</p>
  
  <p>[This seems to be most relevant property to exception.]</p>
  
  <p>[In all cases, driver was set to memory = 51GB, cores = 12, <code>MEMORY_AND_DISK_SER</code> level for caching]</p>
  
  <p>Nothing worked!</p>
  
  <p>If I run the program with half of the bigger dataset size (7.5TB), it finishes successfully in 1.5 hr.</p>
  
  <ol>
  <li>What could I be doing wrong?</li>
  <li>How do I determine the optimal value for <code>fs.s3a.connection.maximum</code>?</li>
  <li>Is it possible that the s3 clients are getting GCed?</li>
  </ol>
  
  <p>Any help will be appreciated!</p>
  
  <p><strong><em>Environment:</em></strong></p>
  
  <p>AWS EMR 5.7.0, 60 x i2.2xlarge SPOT Instances (16 vCPU, 61GB RAM, 2 x 800GB SSD), Spark 2.1.0</p>
  
  <p>YARN is used as resource manager.</p>
  
  <p><strong><em>Code:</em></strong></p>
  
  <p>It's a fairly simple job, doing something like this:</p>
  
  <pre><code>val sl = StorageLevel.MEMORY_AND_DISK_SER
  
  sparkSession.sparkContext.hadoopConfiguration.set("io.compression.codecs", "com.hadoop.compression.lzo.LzopCodec")
  sparkSession.sparkContext.hadoopConfiguration.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
  sparkSession.sparkContext.hadoopConfiguration.setInt("fs.s3a.connection.maximum", 1200)
  
  val dataset_1: DataFrame = sparkSession
      .read
      .format("csv")
      .option("delimiter", ",")
      .schema(&lt;schema: StructType&gt;)
      .csv("s3a://...")
      .select("ID")   //15 TB
  
  dataset_1.persist(sl)
  
  print(dataset_1.count())
  
  tmp = dataset_1.groupBy(“ID”).agg(count("*").alias("count_id”))
  tmp2 = tmp.groupBy("count_id").agg(count("*").alias(“count_count_id”))
  tmp2.write.csv(…)
  
  dataset_1.unpersist()
  </code></pre>
  
  <p><strong><em>Full Stacktrace:</em></strong></p>
  
  <pre><code>17/08/21 20:02:36 INFO compress.CodecPool: Got brand-new decompressor [.lzo]
  17/08/21 20:06:18 WARN lzo.LzopInputStream: IOException in getCompressedData; likely LZO corruption.
  org.apache.http.ConnectionClosedException: Premature end of Content-Length delimited message body (expected: 79627927; received: 19388396
          at org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:180)
          at org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:137)
          at com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:72)
          at com.amazonaws.event.ProgressInputStream.read(ProgressInputStream.java:151)
          at com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:72)
          at com.amazonaws.services.s3.model.S3ObjectInputStream.read(S3ObjectInputStream.java:155)
          at com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:72)
          at com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:72)
          at com.amazonaws.event.ProgressInputStream.read(ProgressInputStream.java:151)
          at com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:72)
          at com.amazonaws.util.LengthCheckInputStream.read(LengthCheckInputStream.java:108)
          at com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:72)
          at com.amazonaws.services.s3.model.S3ObjectInputStream.read(S3ObjectInputStream.java:155)
          at org.apache.hadoop.fs.s3a.S3AInputStream.read(S3AInputStream.java:160)
          at java.io.DataInputStream.read(DataInputStream.java:149)
          at com.hadoop.compression.lzo.LzopInputStream.readFully(LzopInputStream.java:73)
          at com.hadoop.compression.lzo.LzopInputStream.getCompressedData(LzopInputStream.java:321)
          at com.hadoop.compression.lzo.LzopInputStream.decompress(LzopInputStream.java:261)
          at org.apache.hadoop.io.compress.DecompressorStream.read(DecompressorStream.java:85)
          at java.io.InputStream.read(InputStream.java:101)
          at org.apache.hadoop.util.LineReader.fillBuffer(LineReader.java:180)
          at org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:216)
          at org.apache.hadoop.util.LineReader.readLine(LineReader.java:174)
          at org.apache.hadoop.mapreduce.lib.input.LineRecordReader.nextKeyValue(LineRecordReader.java:186)
          at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:39)
          at org.apache.spark.sql.execution.datasources.HadoopFileLinesReader.hasNext(HadoopFileLinesReader.scala:50)
          at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
          at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:461)
          at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
          at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439)
          at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
          at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:109)
          at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
          at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
          at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
          at org.apache.spark.sql.execution.columnar.InMemoryRelation$$anonfun$1$$anon$1.next(InMemoryRelation.scala:99)
          at org.apache.spark.sql.execution.columnar.InMemoryRelation$$anonfun$1$$anon$1.next(InMemoryRelation.scala:91)
          at org.apache.spark.storage.memory.MemoryStore.putIteratorAsBytes(MemoryStore.scala:364)
          at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1021)
          at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:996)
          at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:936)
          at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:996)
          at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:700)
          at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:334)
          at org.apache.spark.rdd.RDD.iterator(RDD.scala:285)
          at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
          at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
  </code></pre>
  
  <p><strong>EDIT:</strong> We have another service which consume exactly same logs, it works just fine. But it uses old "s3://" scheme and is based on Spark-1.6. I'll try using "s3://" instead of "s3a://".</p>