Spark python MLlib 随机森林内存不足错误
Spark python MLlib Random Forest out of memory error
我是运行spark 1.2.1训练随机森林。我在 AWS EC2 上设置了一个主节点和一个工作节点，总共分配了 96GB 的内存给 spark。我尝试了各种并行度值（32、64、6400），但我一直遇到同样的错误。根据 spark UI，我的 RDD 是 277KB，100% 缓存在内存中，应该很小。我的火花上下文如下：
spark.executor.memory 100000m
spark.driver.memory 90000m
spark.driver.maxResultSize 0
spark.storage.memoryFraction 0.6
spark.default.parallelism 6400
spark.eventLog.enabled true
spark.executor.extraLibraryPath /root/ephemeral-hdfs/lib/native/
spark.executor.extraClassPath /root/ephemeral-hdfs/conf
# for spark version < 1.4.0
spark.tachyonStore.url tachyon://10.0.29.29:19998
# for spark version >= 1.4.0
spark.externalBlockStore.url tachyon://10.0.29.29:19998
错误如下：
15/09/16 21:32:38 INFO scheduler.TaskSetManager: Finished task 6397.0 in stage 31.0 (TID 51197) in 4 ms on 10.0.29.186 (6398/6400)
15/09/16 21:32:38 INFO scheduler.TaskSetManager: Finished task 6398.0 in stage 31.0 (TID 51198) in 4 ms on 10.0.29.186 (6399/6400)
15/09/16 21:32:38 INFO scheduler.TaskSetManager: Finished task 6399.0 in stage 31.0 (TID 51199) in 4 ms on 10.0.29.186 (6400/6400)
15/09/16 21:32:38 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 31.0, whose tasks have all completed, from pool
15/09/16 21:32:38 INFO scheduler.DAGScheduler: ResultStage 31 (collect at DecisionTree.scala:977) finished in 1.843 s
15/09/16 21:32:38 INFO scheduler.DAGScheduler: Job 31 finished: collect at DecisionTree.scala:977, took 1.889432 s
15/09/16 21:32:46 INFO storage.BlockManagerInfo: Removed broadcast_32_piece0 on 10.0.29.44:60955 in memory (size: 3.9 KB, free: 45.5 GB)
15/09/16 21:32:46 INFO storage.BlockManagerInfo: Removed broadcast_32_piece0 on 10.0.29.186:43335 in memory (size: 3.9 KB, free: 50.5 GB)
15/09/16 21:34:24 INFO storage.MemoryStore: ensureFreeSpace(46872) called with curMem=0, maxMem=48837427200
15/09/16 21:34:24 INFO storage.MemoryStore: Block broadcast_33 stored as values in memory (estimated size 45.8 KB, free 45.5 GB)
15/09/16 21:34:24 INFO storage.MemoryStore: ensureFreeSpace(46927) called with curMem=46872, maxMem=48837427200
15/09/16 21:34:24 INFO storage.MemoryStore: Block broadcast_33_piece0 stored as bytes in memory (estimated size 45.8 KB, free 45.5 GB)
15/09/16 21:34:24 INFO storage.BlockManagerInfo: Added broadcast_33_piece0 in memory on 10.0.29.44:60955 (size: 45.8 KB, free: 45.5 GB)
15/09/16 21:34:24 INFO spark.SparkContext: Created broadcast 33 from broadcast at DecisionTree.scala:592
15/09/16 21:35:43 INFO rdd.MapPartitionsRDD: Removing RDD 24 from persistence list
15/09/16 21:35:43 INFO storage.BlockManager: Removing RDD 24
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "random_forest_spark.py", line 142, in trainModel
impurity='gini', maxDepth=4, maxBins=32)
File "/root/spark/python/pyspark/mllib/tree.py", line 352, in trainClassifier
maxDepth, maxBins, seed)
File "/root/spark/python/pyspark/mllib/tree.py", line 270, in _train
maxDepth, maxBins, seed)
File "/root/spark/python/pyspark/mllib/common.py", line 128, in callMLlibFunc
return callJavaFunc(sc, api, *args)
File "/root/spark/python/pyspark/mllib/common.py", line 121, in callJavaFunc
return _java2py(sc, func(*args))
File "/root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
File "/root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o169.trainRandomForestModel.
: java.lang.OutOfMemoryError
at java.io.ByteArrayOutputStream.hugeCapacity(ByteArrayOutputStream.java:123)
at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:117)
at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)
at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:153)
at java.io.ObjectOutputStream$BlockDataOutputStream.drain(ObjectOutputStream.java:1876)
at java.io.ObjectOutputStream$BlockDataOutputStream.setBlockDataMode(ObjectOutputStream.java:1785)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1188)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:44)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:81)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:312)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:305)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:132)
at org.apache.spark.SparkContext.clean(SparkContext.scala:1891)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions.apply(RDD.scala:683)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions.apply(RDD.scala:682)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:148)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:109)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:286)
at org.apache.spark.rdd.RDD.mapPartitions(RDD.scala:682)
at org.apache.spark.mllib.tree.DecisionTree$.findBestSplits(DecisionTree.scala:613)
at org.apache.spark.mllib.tree.RandomForest.run(RandomForest.scala:235)
at org.apache.spark.mllib.tree.RandomForest$.trainClassifier(RandomForest.scala:289)
at org.apache.spark.mllib.api.python.PythonMLLibAPI.trainRandomForestModel(PythonMLLibAPI.scala:666)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
at py4j.Gateway.invoke(Gateway.java:259)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:207)
at java.lang.Thread.run(Thread.java:745)
我的数据是 LabeledPoint 类型的 RDD，我的训练代码相当简单：
(trainingData, testData) = data.randomSplit([0.7, 0.3])
model = RandomForest.trainClassifier(trainingData, numClasses=2, categoricalFeaturesInfo={},
numTrees=3, featureSubsetStrategy="auto",
impurity='gini', maxDepth=4, maxBins=32)
原来我的 Tachyon 服务器 IP 不正确。配置正确的 Tachyon 服务器 IP 解决了我的问题。
我是运行spark 1.2.1训练随机森林。我在 AWS EC2 上设置了一个主节点和一个工作节点，总共分配了 96GB 的内存给 spark。我尝试了各种并行度值（32、64、6400），但我一直遇到同样的错误。根据 spark UI，我的 RDD 是 277KB，100% 缓存在内存中，应该很小。我的火花上下文如下：
spark.executor.memory 100000m
spark.driver.memory 90000m
spark.driver.maxResultSize 0
spark.storage.memoryFraction 0.6
spark.default.parallelism 6400
spark.eventLog.enabled true
spark.executor.extraLibraryPath /root/ephemeral-hdfs/lib/native/
spark.executor.extraClassPath /root/ephemeral-hdfs/conf
# for spark version < 1.4.0
spark.tachyonStore.url tachyon://10.0.29.29:19998
# for spark version >= 1.4.0
spark.externalBlockStore.url tachyon://10.0.29.29:19998
错误如下：
15/09/16 21:32:38 INFO scheduler.TaskSetManager: Finished task 6397.0 in stage 31.0 (TID 51197) in 4 ms on 10.0.29.186 (6398/6400)
15/09/16 21:32:38 INFO scheduler.TaskSetManager: Finished task 6398.0 in stage 31.0 (TID 51198) in 4 ms on 10.0.29.186 (6399/6400)
15/09/16 21:32:38 INFO scheduler.TaskSetManager: Finished task 6399.0 in stage 31.0 (TID 51199) in 4 ms on 10.0.29.186 (6400/6400)
15/09/16 21:32:38 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 31.0, whose tasks have all completed, from pool
15/09/16 21:32:38 INFO scheduler.DAGScheduler: ResultStage 31 (collect at DecisionTree.scala:977) finished in 1.843 s
15/09/16 21:32:38 INFO scheduler.DAGScheduler: Job 31 finished: collect at DecisionTree.scala:977, took 1.889432 s
15/09/16 21:32:46 INFO storage.BlockManagerInfo: Removed broadcast_32_piece0 on 10.0.29.44:60955 in memory (size: 3.9 KB, free: 45.5 GB)
15/09/16 21:32:46 INFO storage.BlockManagerInfo: Removed broadcast_32_piece0 on 10.0.29.186:43335 in memory (size: 3.9 KB, free: 50.5 GB)
15/09/16 21:34:24 INFO storage.MemoryStore: ensureFreeSpace(46872) called with curMem=0, maxMem=48837427200
15/09/16 21:34:24 INFO storage.MemoryStore: Block broadcast_33 stored as values in memory (estimated size 45.8 KB, free 45.5 GB)
15/09/16 21:34:24 INFO storage.MemoryStore: ensureFreeSpace(46927) called with curMem=46872, maxMem=48837427200
15/09/16 21:34:24 INFO storage.MemoryStore: Block broadcast_33_piece0 stored as bytes in memory (estimated size 45.8 KB, free 45.5 GB)
15/09/16 21:34:24 INFO storage.BlockManagerInfo: Added broadcast_33_piece0 in memory on 10.0.29.44:60955 (size: 45.8 KB, free: 45.5 GB)
15/09/16 21:34:24 INFO spark.SparkContext: Created broadcast 33 from broadcast at DecisionTree.scala:592
15/09/16 21:35:43 INFO rdd.MapPartitionsRDD: Removing RDD 24 from persistence list
15/09/16 21:35:43 INFO storage.BlockManager: Removing RDD 24
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "random_forest_spark.py", line 142, in trainModel
impurity='gini', maxDepth=4, maxBins=32)
File "/root/spark/python/pyspark/mllib/tree.py", line 352, in trainClassifier
maxDepth, maxBins, seed)
File "/root/spark/python/pyspark/mllib/tree.py", line 270, in _train
maxDepth, maxBins, seed)
File "/root/spark/python/pyspark/mllib/common.py", line 128, in callMLlibFunc
return callJavaFunc(sc, api, *args)
File "/root/spark/python/pyspark/mllib/common.py", line 121, in callJavaFunc
return _java2py(sc, func(*args))
File "/root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
File "/root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o169.trainRandomForestModel.
: java.lang.OutOfMemoryError
at java.io.ByteArrayOutputStream.hugeCapacity(ByteArrayOutputStream.java:123)
at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:117)
at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)
at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:153)
at java.io.ObjectOutputStream$BlockDataOutputStream.drain(ObjectOutputStream.java:1876)
at java.io.ObjectOutputStream$BlockDataOutputStream.setBlockDataMode(ObjectOutputStream.java:1785)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1188)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:44)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:81)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:312)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:305)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:132)
at org.apache.spark.SparkContext.clean(SparkContext.scala:1891)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions.apply(RDD.scala:683)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions.apply(RDD.scala:682)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:148)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:109)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:286)
at org.apache.spark.rdd.RDD.mapPartitions(RDD.scala:682)
at org.apache.spark.mllib.tree.DecisionTree$.findBestSplits(DecisionTree.scala:613)
at org.apache.spark.mllib.tree.RandomForest.run(RandomForest.scala:235)
at org.apache.spark.mllib.tree.RandomForest$.trainClassifier(RandomForest.scala:289)
at org.apache.spark.mllib.api.python.PythonMLLibAPI.trainRandomForestModel(PythonMLLibAPI.scala:666)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
at py4j.Gateway.invoke(Gateway.java:259)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:207)
at java.lang.Thread.run(Thread.java:745)
我的数据是 LabeledPoint 类型的 RDD，我的训练代码相当简单：
(trainingData, testData) = data.randomSplit([0.7, 0.3])
model = RandomForest.trainClassifier(trainingData, numClasses=2, categoricalFeaturesInfo={},
numTrees=3, featureSubsetStrategy="auto",
impurity='gini', maxDepth=4, maxBins=32)
原来我的 Tachyon 服务器 IP 不正确。配置正确的 Tachyon 服务器 IP 解决了我的问题。