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

def submit(self, application='', **kwargs):

    spark_submit_cmd = self._build_command(application)
    self._sp = subprocess.Popen(spark_submit_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=(-1), universal_newlines=True, **kwargs)
    self._process_log(iter(self._sp.stdout.readline, ''))
    returncode = self._sp.wait()
    if returncode:
        raise AirflowException('Cannot    execute:    {}.    Error    code    is:    {}.'.format(spark_submit_cmd, returncode))
