spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Tushar Sudake <etusha...@gmail.com>
Subject 'Premature end of Content-Length' using S3A to read huge data
Date Mon, 21 Aug 2017 21:24:33 GMT
Hello,

I'm writing a Spark based application which works around a pretty huge data
stored on s3. It's about **15 TB** in size uncompressed. Data is laid
across multiple small LZO compressed files files, varying from 10-100MB.

By default the job spawns 130k tasks while reading dataset and mapping it
to schema.

And then it fails around 70k tasks completions and after ~20 tasks failure.

**Exception:**

    WARN lzo.LzopInputStream: IOException in getCompressedData; likely LZO
corruption.
        org.apache.http.ConnectionClosedException: Premature end of
Content-Length delimited message body

Looks like the s3 connection is getting closed prematurely.

**I have tried nearly 40 different combos of configurations.**

**To summarize them:** 1 executor to 3 executors per node, 18GB to 42GB
`--executor-memory`, 3-5 `--executor-cores`, 1.8GB-4.0 GB
`spark.yarn.executor.memoryOverhead`, Both, Kryo and Default Java
serializers, 0.5 to 0.35 `spark.memory.storageFraction`, default, 130000 to
200000 partitions for bigger dataset. default, 200 to 2001
`spark.sql.shuffle.partitions`.

**And most importantly:** 100 to 2048 `fs.s3a.connection.maximum` property.

[This seems to be most relevant property to exception.]

[In all cases, driver was set to memory = 51GB, cores = 12,
`MEMORY_AND_DISK_SER` level for caching]

Nothing worked!

If I run the program with half of the bigger dataset size (7.5TB), it
finishes successfully in 1.5 hr.

1. What could I be doing wrong?
2. How do I determine the optimal value for `fs.s3a.connection.maximum`?
3. Is it possible that the s3 clients are getting GCed?

Any help will be appreciated!

***Environment:***

AWS EMR 5.7.0, 60 x i2.2xlarge SPOT Instances (16 vCPU, 61GB RAM, 2 x 800GB
SSD), Spark 2.1.0

YARN is used as resource manager.

***Code:***

It's a fairly simple job, doing something like this:

    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(<schema: StructType>)
        .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()

***Full Stacktrace:***

    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)

Mime
View raw message