Running Standalone cluster stuck while feeding training data

[Tyron2308]

Hello,

Here's my problem: it stucks while feeding training data.

2018-10-04 04:14:35,233 INFO (MainThread-29426) node: {'executor_id': 1, 'host': '192.168.179.144', 'job_name': 'worker', 'task_index': 0, 'port': 33319, 'tb_pid': 0, 'tb_port': 0, 'addr': '/tmp/pymp-yb30teu7/listener-i7b5ukps', 'authkey': b'뜅�mdOG��1���'}
2018-10-04 04:14:35,233 INFO (MainThread-29426) Starting TensorFlow worker:0 as worker on cluster node 1 on background process
18/10/04 04:14:35 INFO PythonRunner: Times: total = 3479, boot = 479, init = 976, finish = 2024
2018-10-04 04:14:35,245 INFO (MainThread-29466) 1: ======== worker:0 ========
2018-10-04 04:14:35,245 INFO (MainThread-29466) 1: Cluster spec: {'ps': ['192.168.179.144:38999'], 'worker': ['192.168.179.144:33319']}
2018-10-04 04:14:35,245 INFO (MainThread-29466) 1: Using CPU
2018-10-04 04:14:35.246785: I tensorflow/core/platform/cpu_feature_guard.cc:141] Your CPU supports instructions that this TensorFlow binary was not compiled to use: AVX2 FMA
2018-10-04 04:14:35.248870: I tensorflow/core/distributed_runtime/rpc/grpc_channel.cc:222] Initialize GrpcChannelCache for job ps -> {0 -> 192.168.179.144:38999}
2018-10-04 04:14:35.248887: I tensorflow/core/distributed_runtime/rpc/grpc_channel.cc:222] Initialize GrpcChannelCache for job worker -> {0 -> localhost:33319}
2018-10-04 04:14:35.249647: I tensorflow/core/distributed_runtime/rpc/grpc_server_lib.cc:381] Started server with target: grpc://localhost:33319
18/10/04 04:14:35 INFO Executor: Finished task 1.0 in stage 0.0 (TID 1). 1801 bytes result sent to driver
18/10/04 04:14:35 INFO CoarseGrainedExecutorBackend: Got assigned task 2
18/10/04 04:14:35 INFO Executor: Running task 0.0 in stage 1.0 (TID 2)
18/10/04 04:14:35 INFO TorrentBroadcast: Started reading broadcast variable 3
18/10/04 04:14:35 INFO MemoryStore: Block broadcast_3_piece0 stored as bytes in memory (estimated size 7.8 KB, free 366.3 MB)
18/10/04 04:14:35 INFO TorrentBroadcast: Reading broadcast variable 3 took 20 ms
18/10/04 04:14:35 INFO MemoryStore: Block broadcast_3 stored as values in memory (estimated size 14.0 KB, free 366.3 MB)
18/10/04 04:14:35 INFO HadoopRDD: Input split: hdfs://localhost:9000/user/tyron/examples/mnist/csv/train/images/part-00000:0+9338236
18/10/04 04:14:35 INFO TorrentBroadcast: Started reading broadcast variable 0
18/10/04 04:14:35 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 23.0 KB, free 366.2 MB)
18/10/04 04:14:35 INFO TorrentBroadcast: Reading broadcast variable 0 took 19 ms
18/10/04 04:14:35 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 320.9 KB, free 365.9 MB)
tensorflow model path: hdfs://localhost:9000/user/tyron/mnist_model
18/10/04 04:14:36 INFO HadoopRDD: Input split: hdfs://localhost:9000/user/tyron/examples/mnist/csv/train/labels/part-00000:0+204800
18/10/04 04:14:36 INFO TorrentBroadcast: Started reading broadcast variable 1
18/10/04 04:14:36 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 23.0 KB, free 365.9 MB)
18/10/04 04:14:36 INFO TorrentBroadcast: Reading broadcast variable 1 took 13 ms
18/10/04 04:14:36 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 320.9 KB, free 365.6 MB)
2018-10-04 04:14:36,418 INFO (MainThread-29513) Connected to TFSparkNode.mgr on 192.168.179.144, executor=1, state='running'
2018-10-04 04:14:36,432 INFO (MainThread-29513) mgr.state='running'
2018-10-04 04:14:36,432 INFO (MainThread-29513) Feeding partition <itertools.chain object at 0x7f06aa8da048> into input queue <multiprocessing.queues.JoinableQueue object at 0x7f06aa8da358>
18/10/04 04:14:37 INFO PythonRunner: Times: total = 1123, boot = 4, init = 73, finish = 1046
18/10/04 04:14:37 INFO PythonRunner: Times: total = 194, boot = 9, init = 28, finish = 157
18/10/04 04:24:38 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 2)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File /usr/local/lib/spark/python/lib/pyspark.zip/pyspark/worker.py, line 177, in main
process()
File /usr/local/lib/spark/python/lib/pyspark.zip/pyspark/worker.py, line 172, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File /usr/local/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py, line 2423, in pipeline_func
File /usr/local/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py, line 2423, in pipeline_func
File /usr/local/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py, line 2423, in pipeline_func
File /usr/local/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py, line 346, in func
File /usr/local/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py, line 794, in func
File /usr/local/lib/python3.6/dist-packages/tensorflowonspark/TFSparkNode.py, line 414, in _train
raise Exception(Timeout while feeding partition)
Exception: Timeout while feeding partition

at org.apache.spark.api.python.PythonRunner12727anon.read(PythonRDD.scala:193)
at org.apache.spark.api.python.PythonRunner12727anon.<init>(PythonRDD.scala:234)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:108)
at org.apache.spark.executor.Executor.run(Executor.scala:338)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

18/10/04 04:24:38 INFO CoarseGrainedExecutorBackend: Got assigned task 3
18/10/04 04:24:38 INFO Executor: Running task 1.0 in stage 1.0 (TID 3)
18/10/04 04:24:38 INFO HadoopRDD: Input split: hdfs://localhost:9000/user/tyron/examples/mnist/csv/train/images/part-00001:0+11231804
18/10/04 04:24:38 INFO HadoopRDD: Input split: hdfs://localhost:9000/user/tyron/examples/mnist/csv/train/labels/part-00001:0+245760
2018-10-04 04:24:38,940 INFO (MainThread-29986) Connected to TFSparkNode.mgr on 192.168.179.144, executor=1, state='running'
2018-10-04 04:24:38,964 INFO (MainThread-29986) mgr.state='running'
2018-10-04 04:24:38,964 INFO (MainThread-29986) Feeding partition <itertools.chain object at 0x7f06aa8da048> into input queue <multiprocessing.queues.JoinableQueue object at 0x7f06aa8da358>
18/10/04 04:24:40 INFO PythonRunner: Times: total = 1574, boot = 4, init = 31, finish = 1539
18/10/04 04:24:40 INFO PythonRunner: Times: total = 119, boot = 14, init = 52, finish = 53
18/10/04 04:34:41 ERROR Executor: Exception in task 1.0 in stage 1.0 (TID 3)

here are my export :
export LD_LIBRARY_PATH=${PATH}
export HADOOP_HOME=/usr/local/lib/hadoop-2.7.4/
export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop/
export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-amd64
export SPARK_WORKER_INSTANCES=2
export CORES_PER_WORKER=1
export TOTAL_CORES=$((${CORES_PER_WORKER}*${SPARK_WORKER_INSTANCES}))

Here how I run the program :
${SPARK_HOME}/bin/spark-submit
--master ${MASTER}
--py-files ${TFoS_HOME}/examples/mnist/spark/mnist_dist.py
--conf spark.cores.max=${TOTAL_CORES}
--conf spark.task.cpus=${CORES_PER_WORKER}
--conf spark.executorEnv.JAVA_HOME="$JAVA_HOME"
${TFoS_HOME}/examples/mnist/spark/mnist_spark.py
--cluster_size ${SPARK_WORKER_INSTANCES}
--images examples/mnist/csv/train/images
--labels examples/mnist/csv/train/labels
--format csv
--mode train
--model mnist_model

[leewyang]

I noticed that you're using HDFS + Spark Standalone. In that case, you may need the LIB_HDFS settings per the YARN instructions

[Tyron2308]

Hi leewyang,

first of all thank you for answering me.
I added to my bashrc
export LIB_HDFS=$HADOOP_PREFIX/lib/native/Linux-amd64-64
export LIB_JVM=$JAVA_HOME/jre/lib/amd64/server
export QUEUE=default

btw
I can run a simple spark jobs
I can run a simple hdfs request

I still have the same issue but when I add a core to the option --conf spark.cores.max=3 the program stop being stuck but I have the next issue :
#261

Thx

[leewyang]

At the moment, we only support one core per executor

[Tyron2308]

Hi leewyang,

So I tried to print log inside the code to see where exactly and why it was stuck
the execution of the code get stuck when foreachPartition is called.

unionRDD = self.sc.union(rdds)
unionRDD.foreachPartition(TFSparkNode.train(self.cluster_info, self.cluster_meta, feed_timeout=feed_timeout, qname=qname))
then TFSparkNode.train call _train but when this call is made everything get stuck until the exception "timeout while feeding" is raised.
I don't know why but every logs I put inside _train did not log onto the screen.
https://github.com/yahoo/TensorFlowOnSpark/blob/master/tensorflowonspark/TFSparkNode.py#L432

this is how I run the program now :
spark-submit
--master spark://tyron:7077
--queue ${QUEUE}
--py-files examples/mnist/spark/mnist_dist.py
--conf spark.cores.max=2
--conf spark.task.cpus=1
--executor-cores 1
--conf spark.executorENV.JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-amd64/jre/lib/amd64/server --conf spark.executorEnv.CLASSPATH="$($HADOOP_HOME/bin/hadoop classpath --glob)"
--conf spark.executorENV.LD_LIBRARY_PATH=$LIB_HDFS/libhdfs.so examples/mnist/spark/mnist_spark.py
--cluster_size 2 --images examples/mnist/csv/train/images
--labels examples/mnist/csv/train/labels
--format csv --mode train
--model hdfs://user/tyron/mnist_modell

[leewyang]

Can you try the following:

export LIB_HDFS=$HADOOP_PREFIX/lib/native/Linux-amd64-64
export LIB_JVM=$JAVA_HOME/jre/lib/amd64/server

spark-submit
--master spark://tyron:7077
--queue ${QUEUE}
--py-files examples/mnist/spark/mnist_dist.py
--conf spark.cores.max=2
--conf spark.task.cpus=1
--executor-cores 1
--conf spark.executorENV.LD_LIBRARY_PATH=$LIB_HDFS:$LIB_JVM
examples/mnist/spark/mnist_spark.py
--cluster_size 2 --images examples/mnist/csv/train/images
--labels examples/mnist/csv/train/labels
--format csv --mode train
--model hdfs://user/tyron/mnist_modell

If that still hangs, you can look at the Spark UI Executors tab for the job to see the stderr logs of each executor. Hopefully, you will either see the root cause error or at least see the last log statements before the hang.

[Tyron2308]

Hi Leewyang,

Sorry It's me again.. ^^

So I found out where the code was hanging.

joinThr = Thread(target=queue.join)
joinThr.start()
timeout = feed_timeout
while (joinThr.isAlive()):
if (not equeue.empty()): <----------- never goes inside and get the queue
e_str = equeue.get()
equeue.task_done()
raise Exception("exception in worker:\n" + e_str)
time.sleep(1)
timeout -= 1
if timeout <= 0:
raise Exception("Timeout while feeding partition")

So because equeue.empty == True the execution of the program never enter in the if of this part of the code until timeout < 0.
I read your code and understood that you were using the Basemanager TFManager as a shared queue between your workers/ps but I can't find the purpose of equeue and why here :
https://github.com/yahoo/TensorFlowOnSpark/blob/master/tensorflowonspark/TFSparkNode.py#L381 you initialise it with 'error'

[leewyang]

That is a queue to catch errors/exceptions in the TF processes and pass that those exceptions back to Spark (so we can surface them as Spark task failures). So in general, this queue will only be filled if there's a TF exception. That said, I have seen several instances where TF itself can hang w/ no signals to the outside world... If you can send me your logs (leewyang@gmail.com), I can take a closer look...

[manubharghav-zz]

@leewyang , Im facing the same issue as Tyron here. Were you able to resolve their issue ?

[leewyang]

@manubharghav I believe it was tied to not setting the LIB_HDFS env var. Pls post your command line, so I can take a look.

[manubharghav-zz]

this is my command: spark-submit --queue default --master yarn --num-executors 4 --executor-memory 27G --py-files TensorFlowOnSpark/tfspark.zip,TensorFlowOnSpark/examples/mnist/spark/mnist_dist.py --conf spark.dynamicAllocation.enabled=false --conf spark.yarn.maxAppAttempts=1 --conf spark.executorEnv.LD_LIBRARY_PATH=$LIB_JVM:$LIB_HDFS TensorFlowOnSpark/examples/mnist/spark/mnist_spark.py --images hdfs://airfs-silver/user/manu_reddy/mnist_data/train/images --labels hdfs://airfs-silver/user/manu_reddy/mnist_data/train/labels --mode train --model hdfs://airfs-silver/user/manu_reddy/mnist_model

[leewyang]

And, can you confirm that $LIB_JVM points to libjvm.so and $LIB_HDFS points to libhdfs.so on ALL of your grid nodes? And if so, feel free to send me your yarn logs at the email address posted above...

[leewyang]

Closing due to inactivity. Feel free to reopen if still an issue.

[explorerr]

Hello

I am having the same problem, but different situation.

I was able the run the example successfully for one epoch: command as:

hadoop fs -rm -r /rui/misc/mnist/mnist_model 
## Run distributed MNIST training (using feed_dict)
${SPARK_HOME}/bin/spark-submit \
--master yarn \
--deploy-mode cluster \
--num-executors 4 \
--executor-memory 24G \
--py-files TensorFlowOnSpark/examples/mnist/spark/mnist_dist.py \
--conf spark.dynamicAllocation.enabled=false \
--conf spark.yarn.maxAppAttempts=1 \
--conf spark.executorEnv.LD_LIBRARY_PATH=$LIB_JVM:$LIB_HDFS \
TensorFlowOnSpark/examples/mnist/spark/mnist_spark.py \
--images /mnist/csv/train/images \
--labels  /mnist/csv/train/labels \
--mode train \
--model /mnist/mnist_model 

But when I set the epochs to be 100, I got the error saying: Exception: Timeout while feeding partition

hadoop fs -rm -r /rui/misc/mnist/mnist_model 
## Run distributed MNIST training (using feed_dict)
${SPARK_HOME}/bin/spark-submit \
--master yarn \
--deploy-mode cluster \
--num-executors 4 \
--executor-memory 24G \
--py-files TensorFlowOnSpark/examples/mnist/spark/mnist_dist.py \
--conf spark.dynamicAllocation.enabled=false \
--conf spark.yarn.maxAppAttempts=1 \
--conf spark.executorEnv.LD_LIBRARY_PATH=$LIB_JVM:$LIB_HDFS \
TensorFlowOnSpark/examples/mnist/spark/mnist_spark.py \
--images /mnist/csv/train/images \
--labels  /mnist/csv/train/labels \
--mode train \
--model /mnist/mnist_model \
--epochs 100

error message:

19/02/27 13:57:17 WARN TaskSetManager: Lost task 7.0 in stage 1.0 (TID 4, datanode50, executor 1): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/local/hadoop-tmp/nm-local-dir/usercache/hduser/appcache/application_1535736177095_1232825/container_1535736177095_1232825_01_000002/pyspark.zip/pyspark/worker.py", line 230, in main
process()
File "/usr/local/hadoop-tmp/nm-local-dir/usercache/hduser/appcache/application_1535736177095_1232825/container_1535736177095_1232825_01_000002/pyspark.zip/pyspark/worker.py", line 225, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File "/usr/local/hadoop-tmp/nm-local-dir/usercache/hduser/appcache/application_1535736177095_1232825/container_1535736177095_1232825_01_000001/pyspark.zip/pyspark/rdd.py", line 2457, in pipeline_func
File "/usr/local/hadoop-tmp/nm-local-dir/usercache/hduser/appcache/application_1535736177095_1232825/container_1535736177095_1232825_01_000001/pyspark.zip/pyspark/rdd.py", line 2457, in pipeline_func
File "/usr/local/hadoop-tmp/nm-local-dir/usercache/hduser/appcache/application_1535736177095_1232825/container_1535736177095_1232825_01_000001/pyspark.zip/pyspark/rdd.py", line 2457, in pipeline_func
File "/usr/local/hadoop-tmp/nm-local-dir/usercache/hduser/appcache/application_1535736177095_1232825/container_1535736177095_1232825_01_000001/pyspark.zip/pyspark/rdd.py", line 370, in func
File "/usr/local/hadoop-tmp/nm-local-dir/usercache/hduser/appcache/application_1535736177095_1232825/container_1535736177095_1232825_01_000001/pyspark.zip/pyspark/rdd.py", line 819, in func
File "/pairsdatafs03/ipyparallel-software/miniconda/miniconda/envs/py36/lib/python3.6/site-packages/tensorflowonspark/TFSparkNode.py", line 420, in _train
raise Exception("Timeout while feeding partition")
Exception: Timeout while feeding partition

at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1153)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.lang.Thread.run(Thread.java:785)

[leewyang]

@explorerr #400 might be the cause... but in general, for InputMode.SPARK, you really want to closely align the --steps with the --epochs to avoid unnecessary overhead of shipping partitions to the executors when they're not needed.