Catch non-zero exit from pipe commands

megatron-me-uk · megatron-me-uk · commit f552d49127d9 · 2015-05-19T12:01:30.000+01:00
This will allow problems with piped commands to be detected.
This will also allow tasks to be retried where errors are rare (such as network problems in piped commands).
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
@@ -704,7 +704,11 @@ def pipe_objs(out):
                     out.write(s.encode('utf-8'))
                 out.close()
             Thread(target=pipe_objs, args=[pipe.stdin]).start()
-            return (x.rstrip(b'\n').decode('utf-8') for x in iter(pipe.stdout.readline, b''))
+            result = (x.rstrip(b'\n').decode('utf-8') for x in iter(pipe.stdout.readline, b''))
+            pipe.wait()
+            if pipe.returncode:
+                raise Exception("Pipe function `%s' exited with error code %d" %(command, pipe.returncode) )
+            return result
         return self.mapPartitions(func)
 
     def foreach(self, f):
