samedi 8 août 2015

PySpark OSError: [Errno 12] Cannot allocate memory

I've written a binary classifier using Python Theano Library. Based on different data files, I would like to parallelize the classifier based on Amazon Web Service (AWS) EC2 with one master node and several slave nodes using Apache-Spark. When I tested my code with "local" mode on the AWS-EC2 master node which is t2.micro type (1 GB RAM), I got the following error on the memory:

15/08/09 03:57:22 ERROR TaskSetManager: Task 0 in stage 0.0 failed 1 times; aborting job
15/08/09 03:57:22 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
15/08/09 03:57:22 INFO TaskSchedulerImpl: Cancelling stage 0
15/08/09 03:57:22 INFO DAGScheduler: ResultStage 0 (foreach at /root/IdeaNets/Spark/spark_test.py:110) failed in 7.253 s
15/08/09 03:57:22 INFO DAGScheduler: Job 0 failed: foreach at /root/IdeaNets/Spark/spark_test.py:110, took 7.452321 s
Traceback (most recent call last):
  File "/root/IdeaNets/Spark/spark_test.py", line 113, in <module>
    main()
  File "/root/IdeaNets/Spark/spark_test.py", line 110, in main
    datafile.foreach(lambda (path, content): lstm_test(path, content))
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 721, in foreach
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 972, in count
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 963, in sum
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 771, in reduce
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 745, in collect
  File "/root/spark/python/lib/http://ift.tt/1vNTRoR", line 538, in __call__
  File "/root/spark/python/lib/http://ift.tt/10PrSvE", line 300, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/root/spark/python/lib/http://ift.tt/1f113uM", line 111, in main
    process()
  File "/root/spark/python/lib/http://ift.tt/1f113uM", line 106, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 2318, in pipeline_func
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 2318, in pipeline_func
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 2318, in pipeline_func
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 304, in func
  File "/root/spark/python/lib/http://ift.tt/1T7NSuY", line 719, in processPartition
  File "/root/IdeaNets/Spark/spark_test.py", line 110, in <lambda>
    datafile.foreach(lambda (path, content): lstm_test(path, content))
  File "/root/IdeaNets/Spark/spark_test.py", line 71, in lstm_test
    run_lstm.build_model()
  File "/mnt/spark/spark-0b0cd075-92cb-4469-bc85-c347ae6cd58b/userFiles-57a9bb6c-2d93-48fe-a9a4-72d587d70a28/lstm_class.py", line 328, in build_model
    (use_noise, x, mask, y, self.f_pred_prob, self.f_pred, cost) = self._build_model(self._tparams, self.model_options)
  File "/mnt/spark/spark-0b0cd075-92cb-4469-bc85-c347ae6cd58b/userFiles-57a9bb6c-2d93-48fe-a9a4-72d587d70a28/lstm_class.py", line 294, in _build_model
    proj = self.dropout_layer(proj, use_noise, trng)
  File "/mnt/spark/spark-0b0cd075-92cb-4469-bc85-c347ae6cd58b/userFiles-57a9bb6c-2d93-48fe-a9a4-72d587d70a28/lstm_class.py", line 153, in dropout_layer
    dtype=state_before.dtype)),
  File "/usr/local/lib/python2.7/site-packages/theano/sandbox/rng_mrg.py", line 1241, in binomial
    x = self.uniform(size=size, nstreams=nstreams)
  File "/usr/local/lib/python2.7/site-packages/theano/sandbox/rng_mrg.py", line 1219, in uniform
    node_rstate = shared(self.get_substream_rstates(nstreams))
  File "/usr/local/lib/python2.7/site-packages/theano/sandbox/rng_mrg.py", line 1109, in get_substream_rstates
    multMatVect(rval[0], A1p72, M1, A2p72, M2)
  File "/usr/local/lib/python2.7/site-packages/theano/sandbox/rng_mrg.py", line 55, in multMatVect
    [A_sym, s_sym, m_sym, A2_sym, s2_sym, m2_sym], o)
  File "/usr/local/lib/python2.7/site-packages/theano/compile/function.py", line 266, in function
    profile=profile)
  File "/usr/local/lib/python2.7/site-packages/theano/compile/pfunc.py", line 511, in pfunc
    on_unused_input=on_unused_input)
  File "/usr/local/lib/python2.7/site-packages/theano/compile/function_module.py", line 1466, in orig_function
    defaults)
  File "/usr/local/lib/python2.7/site-packages/theano/compile/function_module.py", line 1324, in create
    input_storage=input_storage_lists)
  File "/usr/local/lib/python2.7/site-packages/theano/gof/link.py", line 519, in make_thunk
    output_storage=output_storage)[:3]
  File "/usr/local/lib/python2.7/site-packages/theano/gof/vm.py", line 897, in make_all
    no_recycling))
  File "/usr/local/lib/python2.7/site-packages/theano/gof/op.py", line 739, in make_thunk
    output_storage=node_output_storage)
  File "/usr/local/lib/python2.7/site-packages/theano/gof/cc.py", line 1073, in make_thunk
    keep_lock=keep_lock)
  File "/usr/local/lib/python2.7/site-packages/theano/gof/cc.py", line 1015, in __compile__
    keep_lock=keep_lock)
  File "/usr/local/lib/python2.7/site-packages/theano/gof/cc.py", line 1434, in cthunk_factory
    key = self.cmodule_key()
  File "/usr/local/lib/python2.7/site-packages/theano/gof/cc.py", line 1154, in cmodule_key
    compile_args=self.compile_args(),
  File "/usr/local/lib/python2.7/site-packages/theano/gof/cc.py", line 872, in compile_args
    ret += c_compiler.compile_args()
  File "/usr/local/lib/python2.7/site-packages/theano/gof/cmodule.py", line 1703, in compile_args
    native_lines = get_lines("%s -march=native -E -v -" % theano.config.cxx)
  File "/usr/local/lib/python2.7/site-packages/theano/gof/cmodule.py", line 1672, in get_lines
    shell=True)
  File "/usr/local/lib/python2.7/site-packages/theano/misc/windows.py", line 36, in subprocess_Popen
    proc = subprocess.Popen(command, startupinfo=startupinfo, **params)
  File "/usr/lib64/python2.7/subprocess.py", line 710, in __init__
    errread, errwrite)
  File "/usr/lib64/python2.7/subprocess.py", line 1231, in _execute_child
    self.pid = os.fork()
OSError: [Errno 12] Cannot allocate memory

    at org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:138)
    at org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:179)
    at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:97)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:244)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:63)
    at org.apache.spark.scheduler.Task.run(Task.scala:70)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

Driver stacktrace:
    at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1266)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1257)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1256)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
    at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1256)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:730)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:730)
    at scala.Option.foreach(Option.scala:236)
    at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:730)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1450)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1411)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)

15/08/09 03:57:23 INFO SparkContext: Invoking stop() from shutdown hook
15/08/09 03:57:23 INFO SparkUI: Stopped Spark web UI at http://ift.tt/1T7NQmC
15/08/09 03:57:23 INFO DAGScheduler: Stopping DAGScheduler
15/08/09 03:57:23 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
15/08/09 03:57:23 INFO Utils: path = /mnt/spark/spark-0b0cd075-92cb-4469-bc85-c347ae6cd58b/blockmgr-dc30d0fa-addc-422f-8129-2603af492279, already present as root for deletion.
15/08/09 03:57:23 INFO MemoryStore: MemoryStore cleared
15/08/09 03:57:23 INFO BlockManager: BlockManager stopped
15/08/09 03:57:23 INFO BlockManagerMaster: BlockManagerMaster stopped
15/08/09 03:57:23 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
15/08/09 03:57:23 INFO SparkContext: Successfully stopped SparkContext
15/08/09 03:57:23 INFO Utils: Shutdown hook called
15/08/09 03:57:23 INFO Utils: Deleting directory /mnt/spark/spark-0b0cd075-92cb-4469-bc85-c347ae6cd58b

I'm sure that all my code has no bugs, and it seems that the problem is running out of memory or other problems related to memory. But t2.micro on AWS has 1 GB memory, data that the code reads for training the classifier is just about 1.7 KB, so I think the memory is big enough. I really have no idea about the error as well as how to fix it, and I really appreciate if anyone helps me.




Aucun commentaire:

Enregistrer un commentaire