'Remote Popen (actually execute the Spark-sql query)
:param cmd: command to remotely execute
:param kwargs: extra arguments to Popen (see subprocess.Popen)'

def run_query(self, cmd='', **kwargs):

    prefixed_cmd = self._prepare_command(cmd)
    self._sp = subprocess.Popen(prefixed_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kwargs)
    for line in iter(self._sp.stdout.readline, ''):
        line = line.decode('utf-8').strip()
        logging.info(line)
    for line in iter(self._sp.stderr.readline, ''):
        line = line.decode('utf-8').strip()
        logging.info(line)
    (output, stderr) = self._sp.communicate()
    if self._sp.returncode:
        raise AirflowException('Cannot    execute    {}    on    {}.    Error    code    is:    {}.    Output:    {},    Stderr:    {}'.format(cmd, self._conn.host, self._sp.returncode, output, stderr))
