Catch non-zero exit from pipe commands

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

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