def launch_gateway(conf=None, popen_kwargs=None):
    """
    launch jvm gateway
    :param conf: spark configuration passed to spark-submit
    :param popen_kwargs: Dictionary of kwargs to pass to Popen when spawning
        the py4j JVM. This is a developer feature intended for use in
        customizing how pyspark interacts with the py4j JVM (e.g., capturing
        stdout/stderr).
    :return:
    """
    if "PYSPARK_GATEWAY_PORT" in os.environ:
        gateway_port = int(os.environ["PYSPARK_GATEWAY_PORT"])
        gateway_secret = os.environ["PYSPARK_GATEWAY_SECRET"]
        # Process already exists
        proc = None
    else:
        SPARK_HOME = _find_spark_home()
        # Launch the Py4j gateway using Spark's run command so that we pick up the
        # proper classpath and settings from spark-env.sh
        on_windows = platform.system() == "Windows"
        script = "./bin/spark-submit.cmd" if on_windows else "./bin/spark-submit"
        command = [os.path.join(SPARK_HOME, script)]
        if conf:
            for k, v in conf.getAll():
                command += ['--conf', '%s=%s' % (k, v)]
        submit_args = os.environ.get("PYSPARK_SUBMIT_ARGS", "pyspark-shell")
        if os.environ.get("SPARK_TESTING"):
            submit_args = ' '.join([
                "--conf spark.ui.enabled=false",
                submit_args
            ])
        command = command + shlex.split(submit_args)

        # Create a temporary directory where the gateway server should write the connection
        # information.
        conn_info_dir = tempfile.mkdtemp()
        try:
            fd, conn_info_file = tempfile.mkstemp(dir=conn_info_dir)
            os.close(fd)
            os.unlink(conn_info_file)

            env = dict(os.environ)
            env["_PYSPARK_DRIVER_CONN_INFO_PATH"] = conn_info_file

            # Launch the Java gateway.
            popen_kwargs = {} if popen_kwargs is None else popen_kwargs
            # We open a pipe to stdin so that the Java gateway can die when the pipe is broken
            popen_kwargs['stdin'] = PIPE
            # We always set the necessary environment variables.
            popen_kwargs['env'] = env
            if not on_windows:
                # Don't send ctrl-c / SIGINT to the Java gateway:
                def preexec_func():
                    signal.signal(signal.SIGINT, signal.SIG_IGN)
                popen_kwargs['preexec_fn'] = preexec_func
                proc = Popen(command, **popen_kwargs)
            else:
                # preexec_fn not supported on Windows
                proc = Popen(command, **popen_kwargs)

            # Wait for the file to appear, or for the process to exit, whichever happens first.
            while not proc.poll() and not os.path.isfile(conn_info_file):
                time.sleep(0.1)

            if not os.path.isfile(conn_info_file):
                raise Exception("Java gateway process exited before sending its port number")

            with open(conn_info_file, "rb") as info:
                gateway_port = read_int(info)
                gateway_secret = UTF8Deserializer().loads(info)
        finally:
            shutil.rmtree(conn_info_dir)

        # In Windows, ensure the Java child processes do not linger after Python has exited.
        # In UNIX-based systems, the child process can kill itself on broken pipe (i.e. when
        # the parent process' stdin sends an EOF). In Windows, however, this is not possible
        # because java.lang.Process reads directly from the parent process' stdin, contending
        # with any opportunity to read an EOF from the parent. Note that this is only best
        # effort and will not take effect if the python process is violently terminated.
        if on_windows:
            # In Windows, the child process here is "spark-submit.cmd", not the JVM itself
            # (because the UNIX "exec" command is not available). This means we cannot simply
            # call proc.kill(), which kills only the "spark-submit.cmd" process but not the
            # JVMs. Instead, we use "taskkill" with the tree-kill option "/t" to terminate all
            # child processes in the tree (http://technet.microsoft.com/en-us/library/bb491009.aspx)
            def killChild():
                Popen(["cmd", "/c", "taskkill", "/f", "/t", "/pid", str(proc.pid)])
            atexit.register(killChild)

    # Connect to the gateway
    gateway = JavaGateway(
        gateway_parameters=GatewayParameters(port=gateway_port, auth_token=gateway_secret,
                                             auto_convert=True))
    # Store a reference to the Popen object for use by the caller (e.g., in reading stdout/stderr)
    gateway.proc = proc

    # Import the classes used by PySpark
    java_import(gateway.jvm, "org.apache.spark.SparkConf")
    java_import(gateway.jvm, "org.apache.spark.api.java.*")
    java_import(gateway.jvm, "org.apache.spark.api.python.*")
    java_import(gateway.jvm, "org.apache.spark.ml.python.*")
    java_import(gateway.jvm, "org.apache.spark.mllib.api.python.*")
    # TODO(davies): move into sql
    java_import(gateway.jvm, "org.apache.spark.sql.*")
    java_import(gateway.jvm, "org.apache.spark.sql.api.python.*")
    java_import(gateway.jvm, "org.apache.spark.sql.hive.*")
    java_import(gateway.jvm, "scala.Tuple2")

    return gateway
def launch_gateway ( conf = None , popen_kwargs = None ) : if "PYSPARK_GATEWAY_PORT" in os . environ : gateway_port = int ( os . environ [ "PYSPARK_GATEWAY_PORT" ] ) gateway_secret = os . environ [ "PYSPARK_GATEWAY_SECRET" ] proc = None else : SPARK_HOME = _find_spark_home ( ) on_windows = platform . system ( ) == "Windows" script = "./bin/spark-submit.cmd" if on_windows else "./bin/spark-submit" command = [ os . path . join ( SPARK_HOME , script ) ] if conf : for k , v in conf . getAll ( ) : command += [ '--conf' , '%s=%s' % ( k , v ) ] submit_args = os . environ . get ( "PYSPARK_SUBMIT_ARGS" , "pyspark-shell" ) if os . environ . get ( "SPARK_TESTING" ) : submit_args = ' ' . join ( [ "--conf spark.ui.enabled=false" , submit_args ] ) command = command + shlex . split ( submit_args ) conn_info_dir = tempfile . mkdtemp ( ) try : fd , conn_info_file = tempfile . mkstemp ( dir = conn_info_dir ) os . close ( fd ) os . unlink ( conn_info_file ) env = dict ( os . environ ) env [ "_PYSPARK_DRIVER_CONN_INFO_PATH" ] = conn_info_file popen_kwargs = { } if popen_kwargs is None else popen_kwargs popen_kwargs [ 'stdin' ] = PIPE popen_kwargs [ 'env' ] = env if not on_windows : def preexec_func ( ) : signal . signal ( signal . SIGINT , signal . SIG_IGN ) popen_kwargs [ 'preexec_fn' ] = preexec_func proc = Popen ( command , ** popen_kwargs ) else : proc = Popen ( command , ** popen_kwargs ) while not proc . poll ( ) and not os . path . isfile ( conn_info_file ) : time . sleep ( 0.1 ) if not os . path . isfile ( conn_info_file ) : raise Exception ( "Java gateway process exited before sending its port number" ) with open ( conn_info_file , "rb" ) as info : gateway_port = read_int ( info ) gateway_secret = UTF8Deserializer ( ) . loads ( info ) finally : shutil . rmtree ( conn_info_dir ) if on_windows : def killChild ( ) : Popen ( [ "cmd" , "/c" , "taskkill" , "/f" , "/t" , "/pid" , str ( proc . pid ) ] ) atexit . register ( killChild ) gateway = JavaGateway ( gateway_parameters = GatewayParameters ( port = gateway_port , auth_token = gateway_secret , auto_convert = True ) ) gateway . proc = proc java_import ( gateway . jvm , "org.apache.spark.SparkConf" ) java_import ( gateway . jvm , "org.apache.spark.api.java.*" ) java_import ( gateway . jvm , "org.apache.spark.api.python.*" ) java_import ( gateway . jvm , "org.apache.spark.ml.python.*" ) java_import ( gateway . jvm , "org.apache.spark.mllib.api.python.*" ) java_import ( gateway . jvm , "org.apache.spark.sql.*" ) java_import ( gateway . jvm , "org.apache.spark.sql.api.python.*" ) java_import ( gateway . jvm , "org.apache.spark.sql.hive.*" ) java_import ( gateway . jvm , "scala.Tuple2" ) return gateway
launch jvm gateway
    :param conf: spark configuration passed to spark-submit
    :param popen_kwargs: Dictionary of kwargs to pass to Popen when spawning
        the py4j JVM. This is a developer feature intended for use in
        customizing how pyspark interacts with the py4j JVM (e.g., capturing
        stdout/stderr).
    :return: