airflow/contrib/hooks/spark_submit_hook.py
def submit(self, application="", **kwargs):
        """
        Remote Popen to execute the spark-submit job

        :param application: Submitted application, jar or py file
        :type application: str
        :param kwargs: extra arguments to Popen (see subprocess.Popen)
        """
        spark_submit_cmd = self._build_spark_submit_command(application)

        if hasattr(self, '_env'):
            env = os.environ.copy()
            env.update(self._env)
            kwargs["env"] = env

        self._submit_sp = subprocess.Popen(spark_submit_cmd,
                                           stdout=subprocess.PIPE,
                                           stderr=subprocess.STDOUT,
                                           bufsize=-1,
                                           universal_newlines=True,
                                           **kwargs)

        self._process_spark_submit_log(iter(self._submit_sp.stdout.readline, ''))
        returncode = self._submit_sp.wait()

        # Check spark-submit return code. In Kubernetes mode, also check the value
        # of exit code in the log, as it may differ.
        if returncode or (self._is_kubernetes and self._spark_exit_code != 0):
            raise AirflowException(
                "Cannot execute: {}. Error code is: {}.".format(
                    spark_submit_cmd, returncode
                )
            )

        self.log.debug("Should track driver: {}".format(self._should_track_driver_status))

        # We want the Airflow job to wait until the Spark driver is finished
        if self._should_track_driver_status:
            if self._driver_id is None:
                raise AirflowException(
                    "No driver id is known: something went wrong when executing " +
                    "the spark submit command"
                )

            # We start with the SUBMITTED status as initial status
            self._driver_status = "SUBMITTED"

            # Start tracking the driver status (blocking function)
            self._start_driver_status_tracking()

            if self._driver_status != "FINISHED":
                raise AirflowException(
                    "ERROR : Driver {} badly exited with status {}"
                    .format(self._driver_id, self._driver_status)
                )
def submit(self, application="", **kwargs):
        spark_submit_cmd = self._build_spark_submit_command(application)

        if hasattr(self, '_env'):
            env = os.environ.copy()
            env.update(self._env)
            kwargs["env"] = env

        self._submit_sp = subprocess.Popen(spark_submit_cmd,
                                           stdout=subprocess.PIPE,
                                           stderr=subprocess.STDOUT,
                                           bufsize=-1,
                                           universal_newlines=True,
                                           **kwargs)

        self._process_spark_submit_log(iter(self._submit_sp.stdout.readline, ''))
        returncode = self._submit_sp.wait()

        # Check spark-submit return code. In Kubernetes mode, also check the value
        # of exit code in the log, as it may differ.
        if returncode or (self._is_kubernetes and self._spark_exit_code != 0):
            raise AirflowException(
                "Cannot execute: {}. Error code is: {}.".format(
                    spark_submit_cmd, returncode
                )
            )

        self.log.debug("Should track driver: {}".format(self._should_track_driver_status))

        # We want the Airflow job to wait until the Spark driver is finished
        if self._should_track_driver_status:
            if self._driver_id is None:
                raise AirflowException(
                    "No driver id is known: something went wrong when executing " +
                    "the spark submit command"
                )

            # We start with the SUBMITTED status as initial status
            self._driver_status = "SUBMITTED"

            # Start tracking the driver status (blocking function)
            self._start_driver_status_tracking()

            if self._driver_status != "FINISHED":
                raise AirflowException(
                    "ERROR : Driver {} badly exited with status {}"
                    .format(self._driver_id, self._driver_status)
                )
def submit ( self , application = "" , * * kwargs ) : spark_submit_cmd = self . _build_spark_submit_command ( application ) if hasattr ( self , '_env' ) : env = os . environ . copy ( ) env . update ( self . _env ) kwargs [ "env" ] = env self . _submit_sp = subprocess . Popen ( spark_submit_cmd , stdout = subprocess . PIPE , stderr = subprocess . STDOUT , bufsize = - 1 , universal_newlines = True , * * kwargs ) self . _process_spark_submit_log ( iter ( self . _submit_sp . stdout . readline , '' ) ) returncode = self . _submit_sp . wait ( ) # Check spark-submit return code. In Kubernetes mode, also check the value # of exit code in the log, as it may differ. if returncode or ( self . _is_kubernetes and self . _spark_exit_code != 0 ) : raise AirflowException ( "Cannot execute: {}. Error code is: {}." . format ( spark_submit_cmd , returncode ) ) self . log . debug ( "Should track driver: {}" . format ( self . _should_track_driver_status ) ) # We want the Airflow job to wait until the Spark driver is finished if self . _should_track_driver_status : if self . _driver_id is None : raise AirflowException ( "No driver id is known: something went wrong when executing " + "the spark submit command" ) # We start with the SUBMITTED status as initial status self . _driver_status = "SUBMITTED" # Start tracking the driver status (blocking function) self . _start_driver_status_tracking ( ) if self . _driver_status != "FINISHED" : raise AirflowException ( "ERROR : Driver {} badly exited with status {}" . format ( self . _driver_id , self . _driver_status ) )
Remote Popen to execute the spark-submit job

        :param application: Submitted application, jar or py file
        :type application: str
        :param kwargs: extra arguments to Popen (see subprocess.Popen)
Remote Popen to execute the spark - submit job