CODEWITHSUNDEEP
javahadoopapache-sparkamazon-s3
Ganesh R
Ganesh R

Reputation: 123

SPARK driver running out of memory while reading several S3 files

Situation

I am new to SPARK, I am running a SPARK job in EMR which reads a bunch of S3 files and and performs Map/reduce jobs. There are a total of 200 S3 locations which on a average contains 400 files.

The textFile(...) API is called with a comma separated S3 path with a wildcard (*)in the end example: 

sc.textFile("S3://FilePath1/\*","S3://FilePath2/\*"....."S3://FilePath200/\*")

The job spends a lot of time in the Driver, and finally runs out of memory with the below error. 

Container [pid=66583,containerID=container_1507231957101_0001_02_000001] is running beyond physical memory limits. 
Current usage: 1.5 GB of 1.4 GB physical memory used; 3.3 GB of 6.9 GB virtual memory used. Killing container.
Dump of the process-tree for container_1507231957101_0001_02_000001

Questions

  1. I am setting the driver memory to 32g using the below code, but the driver is still running with 1.4g. Am I missing something ? I am submitting the job with spark-submit --verbose --deploy-mode cluster
private void initializeSparkContext() {
    final SparkConf conf = new SparkConf().setAppName(comparisonJobArgument.getAppName());
    conf.set("spark.driver.memory", "32g");
    conf.set("spark.files.maxPartitionBytes", "134217728");
    context = new JavaSparkContext(conf);
}

Adding more code

RDD1 

context
    .textFile(commaSeperatedS3Locations) // 200 folder like s3://path/* with 400 items in each folder
    .mapPartitions(StringToObjectTransformer())
    .filter(filter)

RDD2 

context
  .textFile(commaSeperatedS3Locations) // 1280 s3 files
  .mapPartitions(StringToObjectTransformer())
  .filter(filter)
  .map(Object1ToObject2Transformer())
  .flatMap(k -> k.iterator())

RDD3 

context.union(RDD1)
  .union(RDD2)
  .map(Object1ToObject2Transformer)
  .mapToPair(mapToPairObject)
  .reduceByKey()
  .coalase(320,false)
  .cache(); // I have total of 1TB executor memory.

And the saveAsTextFile statements : 

RDD3.filter(filter1).saveToTextFile(s3://OutputPath1);
RDD3.filter(filter2).saveToTextFile(s3://OutputPath2);
RDD3.filter(filter3).saveToTextFile(s3://OutputPath3);
RDD3.filter(filter4).saveToTextFile(s3://OutputPath4);
RDD3.filter(filter5).saveToTextFile(s3://OutputPath5);

Greatly appreciate your help on this.

Thanks in advance.

Full ERROR Message

Application application_1507231957101_0001 failed 2 times due to AM Container for appattempt_1507231957101_0001_000002 exited with exitCode: -104
For more detailed output, check application tracking page:http://ip-172-16-0-98.us-west-2.compute.internal:8088/cluster/app/application_1507231957101_0001Then, click on links to logs of each attempt.
**Diagnostics: Container [pid=66583,containerID=container_1507231957101_0001_02_000001] is running beyond physical memory limits. Current usage: 1.5 GB of 1.4 GB physical memory used; 3.3 GB of 6.9 GB virtual memory used. Killing container.***
Dump of the process-tree for container_1507231957101_0001_02_000001 :
|- PID PPID PGRPID SESSID CMD_NAME USER_MODE_TIME(MILLIS) SYSTEM_TIME(MILLIS) VMEM_USAGE(BYTES) RSSMEM_USAGE(PAGES) FULL_CMD_LINE
|- 66583 66581 66583 66583 (bash) 0 0 115814400 688 /bin/bash -c LD_LIBRARY_PATH=/usr/lib/hadoop/lib/native:/usr/lib/hadoop-lzo/lib/native:::/usr/lib/hadoop-lzo/lib/native:/usr/lib/hadoop/lib/native::/usr/lib/hadoop-lzo/lib/native:/usr/lib/hadoop/lib/native:/usr/lib/hadoop-lzo/lib/native:/usr/lib/hadoop/lib/native /usr/lib/jvm/java-openjdk/bin/java -server -Xmx1024m -Djava.io.tmpdir=/mnt/yarn/usercache/hadoop/appcache/application_1507231957101_0001/container_1507231957101_0001_02_000001/tmp '-XX:+UseConcMarkSweepGC' '-XX:CMSInitiatingOccupancyFraction=70' '-XX:MaxHeapFreeRatio=70' '-XX:+CMSClassUnloadingEnabled' '-XX:OnOutOfMemoryError=kill -9 %p' -Dspark.yarn.app.container.log.dir=/var/log/hadoop-yarn/containers/application_1507231957101_0001/container_1507231957101_0001_02_000001 org.apache.spark.deploy.yarn.ApplicationMaster --class 'com.amazon.reconcentral.comparisonengine.jobs.main.ComparisonJob' --jar s3://recon-central-test-usamazon/lib/comparison-engine/ReconCentralComparisonEngine-1.0-super.jar --arg '-s3B' --arg 'recon-central-test-usamazon' --arg '-s3L' --arg 'var/args/comparison-engine/ComparisonEngine:RC_ACETOUSL_ALLREGION.750.bWcQFMA.301-d25518a5-459e-49f0-8d6b-71ad695bbb7f.json' --arg '-s3E' --arg '3ebfb91d-faf0-4295-a5d9-408080e71841' --properties-file /mnt/yarn/usercache/hadoop/appcache/application_1507231957101_0001/container_1507231957101_0001_02_000001/__spark_conf__/__spark_conf__.properties 1> /var/log/hadoop-yarn/containers/application_1507231957101_0001/container_1507231957101_0001_02_000001/stdout 2> /var/log/hadoop-yarn/containers/application_1507231957101_0001/container_1507231957101_0001_02_000001/stderr
|- 66588 66583 66583 66583 (java) 27893 936 3445600256 385188 /usr/lib/jvm/java-openjdk/bin/java -server -Xmx1024m -Djava.io.tmpdir=/mnt/yarn/usercache/hadoop/appcache/application_1507231957101_0001/container_1507231957101_0001_02_000001/tmp -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=70 -XX:MaxHeapFreeRatio=70 -XX:+CMSClassUnloadingEnabled -XX:OnOutOfMemoryError=kill -9 %p -Dspark.yarn.app.container.log.dir=/var/log/hadoop-yarn/containers/application_1507231957101_0001/container_1507231957101_0001_02_000001 org.apache.spark.deploy.yarn.ApplicationMaster --class com.amazon.reconcentral.comparisonengine.jobs.main.ComparisonJob --jar s3://recon-central-test-usamazon/lib/comparison-engine/ReconCentralComparisonEngine-1.0-super.jar --arg -s3B --arg recon-central-test-usamazon --arg -s3L --arg var/args/comparison-engine/ComparisonEngine:RC_ACETOUSL_ALLREGION.750.bWcQFMA.301-d25518a5-459e-49f0-8d6b-71ad695bbb7f.json --arg -s3E --arg 3ebfb91d-faf0-4295-a5d9-408080e71841 --properties-file /mnt/yarn/usercache/hadoop/appcache/application_1507231957101_0001/container_1507231957101_0001_02_000001/__spark_conf__/__spark_conf__.properties
Container killed on request. Exit code is 143
Container exited with a non-zero exit code 143
Failing this attempt. Failing the application.

Upvotes: 1

Views: 2280

Answers (1)

vaquar khan
vaquar khan

Reputation: 11449

SPARK_WORKER_MEMORY is only used in standalone deploy mode

SPARK_EXECUTOR_MEMORY is used in YARN deploy mode

You can either launch your spark-shell using:

./bin/spark-shell --driver-memory 40g

or

You can set it in spark-defaults.conf:

spark.driver.memory 40g

or

If you are launching an application using spark-submit, you must specify the driver memory as an argument:

./bin/spark-submit --driver-memory 40g --class main.class yourApp.jar

Properties set directly on the SparkConf take highest precedence, then flags passed to spark-submit or spark-shell, then options in the spark-defaults.conf file.

This is the priority order (from highest to lowest):

  1. Properties set on the SparkConf (in program). Flags passed to
  2. spark-submit or spark-shell.
  3. Options set in the spark-defaults.conf file.

Run on a Spark standalone cluster in client deploy mode

./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://207.184.161.138:7077 \
  --executor-memory 20G \
  --total-executor-cores 100 \
  /path/to/examples.jar \
  1000

Run on a Spark standalone cluster in cluster deploy mode with supervise

./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://207.184.161.138:7077 \
  --deploy-mode cluster \
  --supervise \
  --executor-memory 20G \
  --total-executor-cores 100 \
  /path/to/examples.jar \
  1000

Run on a YARN cluster

export HADOOP_CONF_DIR=XXX
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn \
  --deploy-mode cluster \  # can be client for client mode
  --executor-memory 20G \
  --num-executors 50 \
  /path/to/examples.jar \
  1000

Run a Python application on a Spark standalone cluster

./bin/spark-submit \
  --master spark://207.184.161.138:7077 \
  examples/src/main/python/pi.py \
  1000

Run on a Mesos cluster in cluster deploy mode with supervise

./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master mesos://207.184.161.138:7077 \
  --deploy-mode cluster \
  --supervise \
  --executor-memory 20G \
  --total-executor-cores 100 \
  http://path/to/examples.jar \
  1000

*http://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark-configure.html

Upvotes: 2

Related Questions