I encounter big problems with my spark application in terms of memory usage. So I'm running spark on an aws emr cluster which in cluster mode(which means it uses yarn as cluster manager). I can't give any code of the application, because it's big and I don't know where the problem is but basically what I do is to load several types of json objects from aws s3(the data is stored in different folders by hour, i.e 2017/05/31/00 is the data which should be processed by the spark application for 00:00 on the 31 of May 2017, and I run the application for every hour), deserialize them to Java pojo objects, made some transformations and cache them because I don't want to reload them on every action. Some of the objects are transformed as a java map and after all logic is completed the java map is transformed back to a RDD and saved back in s3.
Problem 1:
With the following configurations:
-conf spark.yarn.executor.memoryOverhead=3000 --driver-memory 8G --executor-memory 8G --deploy-mode cluster
The Map which I transformed to an RDD and save it is a one which size with is increased with every executed hour. But on completely randomly on some executions the saved output has missing data. For example if the count of json objects after a successful execution is 1000 and on the next it's possible to make 900 instead of being increased. And if you run it again on the same hour it is absolutely possible to be ok(I can't see a bug in my application as this is happens extremely rarely and randomly). So that makes me think that it's a memory issue and maybe some of the objects in the memory are garbage collected or something like that.
Problem 2:
When I run the program with the following configurations:
-conf spark.yarn.executor.memoryOverhead=3000 --driver-memory 8G --executor-memory 1G --deploy-mode cluster
for several hours in one execution(doing a while loop for every hour within the application) then everything is fine. But if I change the executor memory to more than 1G on the third hour a really strange exception is thrown:
17/05/31 09:38:52 INFO S3NativeFileSystem: Encountered an exception while reading 'data/appletv/user_mapping/part-00000.gz', will retry by attempting to reopen stream.
java.io.EOFException: Invalid position: 1856, exceeds the bounds of the stream: [0, 1855]
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem$NativeS3FsInputStream.throwPositionOutOfBoundsException(S3NativeFileSystem.java:254)
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem$NativeS3FsInputStream.retrievePair(S3NativeFileSystem.java:238)
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem$NativeS3FsInputStream.reopenStream(S3NativeFileSystem.java:291)
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem$NativeS3FsInputStream.read(S3NativeFileSystem.java:187)
at java.io.BufferedInputStream.read1(BufferedInputStream.java:284)
at java.io.BufferedInputStream.read(BufferedInputStream.java:345)
at java.io.DataInputStream.read(DataInputStream.java:149)
at org.apache.hadoop.io.compress.DecompressorStream.getCompressedData(DecompressorStream.java:159)
at org.apache.hadoop.io.compress.DecompressorStream.decompress(DecompressorStream.java:143)
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.rdd.NewHadoopRDD$$anon$1.hasNext(NewHadoopRDD.scala:199)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:149)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
at org.apache.spark.scheduler.Task.run(Task.scala:99)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
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)
I really can't fin a logical explanation of these two problems. It would be nice if someone can help.