java - SPARK 驱动程序在读取多个 S3 文件时内存不足

Question

情况

我是 SPARK 的新手，我在 EMR 中运行 SPARK 作业，它读取一堆 S3 文件并执行 Map/reduce 作业。总共有 200 个 S3 位置，平均包含 400 个文件。

在最后的示例中，textFile(...) API 使用逗号分隔的 S3 路径和通配符 (*) 调用:

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

该作业在驱动程序中花费了大量时间，最终内存不足并出现以下错误。

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

问题

1. 我使用以下代码将驱动程序内存设置为 32g，但驱动程序仍在以 1.4g 运行。我错过了什么吗？我正在使用 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);
}

添加更多代码

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.

和 saveAsTextFile 语句:

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);

非常感谢您对此提供的帮助。

提前致谢。

完整错误信息

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.

Answer 1

SPARK_WORKER_MEMORY is only used in standalone deploy mode

SPARK_EXECUTOR_MEMORY is used in YARN deploy mode

您可以使用以下方式启动您的 spark-shell:

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

或

可以在spark-defaults.conf中设置:

spark.driver.memory 40g

或

如果您使用 spark-submit 启动应用程序，则必须将驱动程序内存指定为参数:

./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.

这是优先级顺序(从最高到最低):

1. 在 SparkConf 上设置的属性(在程序中)。标志传递给
2. spark-submit 或 spark-shell。
3. 在 spark-defaults.conf 中设置的选项 文件。

以客户端部署模式在 Spark 独立集群上运行

./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

在带有监督的集群部署模式下在 Spark 独立集群上运行

./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

在 YARN 集群上运行

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

在 Spark 独立集群上运行 Python 应用程序

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

在带有监督的集群部署模式下在 Mesos 集群上运行

./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

• https://spark.apache.org/docs/latest/submitting-applications.html

• https://spark.apache.org/docs/latest/tuning.html