python/pyspark/rdd.py
def pipe(self, command, env={}):
        """
        Return an RDD created by piping elements to a forked external process.

        >>> sc.parallelize([1, 2, 3]).pipe('cat').collect()
        ['1', '2', '3']
        """
        def func(iterator):
            pipe = Popen(shlex.split(command), env=env, stdin=PIPE, stdout=PIPE)
            def pipe_objs(out):
                for obj in iterator:
                    out.write(str(obj).rstrip('\n') + '\n')
                out.close()
            Thread(target=pipe_objs, args=[pipe.stdin]).start()
            return (x.rstrip('\n') for x in pipe.stdout)
        return self.mapPartitions(func)

        Return an RDD created by piping elements to a forked external process.

        >>> sc.parallelize([1, 2, 3]).pipe('cat').collect()
        ['1', '2', '3']
        
Return an RDD created by piping elements to a forked external process.
Return an RDD created by piping elements to a forked external process .
def pipe(self, command, env={}):
        def func(iterator):
            pipe = Popen(shlex.split(command), env=env, stdin=PIPE, stdout=PIPE)
            def pipe_objs(out):
                for obj in iterator:
                    out.write(str(obj).rstrip('\n') + '\n')
                out.close()
            Thread(target=pipe_objs, args=[pipe.stdin]).start()
            return (x.rstrip('\n') for x in pipe.stdout)
        return self.mapPartitions(func)
def pipe ( self , command , env = { } ) : def func ( iterator ) : pipe = Popen ( shlex . split ( command ) , env = env , stdin = PIPE , stdout = PIPE ) def pipe_objs ( out ) : for obj in iterator : out . write ( str ( obj ) . rstrip ( '\n' ) + '\n' ) out . close ( ) Thread ( target = pipe_objs , args = [ pipe . stdin ] ) . start ( ) return ( x . rstrip ( '\n' ) for x in pipe . stdout ) return self . mapPartitions ( func )
Return an RDD created by piping elements to a forked external process.
Return an RDD created by piping elements to a forked external process .
"""
        Return an RDD created by piping elements to a forked external process.

        >>> sc.parallelize([1, 2, 3]).pipe('cat').collect()
        ['1', '2', '3']
        """
self
command
env