'Run an hql statement using the hive cli. If hive_conf is specified
it should be a dict and the entries will be set as key/value pairs
in HiveConf
:param hive_conf: if specified these key value pairs will be passed
to hive as ``-hiveconf "key"="value"``. Note that they will be
passed after the ``hive_cli_params`` and thus will override
whatever values are specified in the database.
:type hive_conf: dict
>>> hh = HiveCliHook()
>>> result = hh.run_cli("USE airflow;")
>>> ("OK" in result)
True'

def run_cli(self, hql, schema=None, verbose=True, hive_conf=None):

    conn = self.conn
    schema = (schema or conn.schema)
    if schema:
        hql = 'USE    {schema};\n{hql}'.format(**locals())
    with TemporaryDirectory(prefix='airflow_hiveop_') as tmp_dir:
        with NamedTemporaryFile(dir=tmp_dir) as f:
            f.write(hql.encode('UTF-8'))
            f.flush()
            hive_cmd = self._prepare_cli_cmd()
            hive_conf_params = self._prepare_hiveconf(hive_conf)
            if self.mapred_queue:
                hive_conf_params.extend(['-hiveconf', 'mapreduce.job.queuename={}'.format(self.mapred_queue)])
            if self.mapred_queue_priority:
                hive_conf_params.extend(['-hiveconf', 'mapreduce.job.priority={}'.format(self.mapred_queue_priority)])
            if self.mapred_job_name:
                hive_conf_params.extend(['-hiveconf', 'mapred.job.name={}'.format(self.mapred_job_name)])
            hive_cmd.extend(hive_conf_params)
            hive_cmd.extend(['-f', f.name])
            if verbose:
                logging.info('    '.join(hive_cmd))
            sp = subprocess.Popen(hive_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, cwd=tmp_dir)
            self.sp = sp
            stdout = ''
            while True:
                line = sp.stdout.readline()
                if (not line):
                    break
                stdout += line.decode('UTF-8')
                if verbose:
                    logging.info(line.decode('UTF-8').strip())
            sp.wait()
            if sp.returncode:
                raise AirflowException(stdout)
            return stdout
