如何使用 Python 将 Jinja2 传递给 HiveQL

Question

我使用 Gcloud Composer 作为我的 Airflow。当我尝试在我的 HQL 代码中使用 Jinja 时，它没有正确翻译它。 我知道 HiveOperator 有一个 Jinja 翻译器，因为我已经习惯了，但 DataProcHiveOperator 没有。

我尝试将 HiveConf 直接用于我的 HQL 文件，但是在将这些值设置为我的分区（即INSERT INTO TABLE abc PARTITION (ds = ${hiveconf:ds})）时，它不起作用。 我还在我的 HQL 文件中添加了以下内容：

SET ds=to_date(current_timestamp());

SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;

但它不起作用，因为 HIVE 正在将上面的公式转换为字符串。

所以我的想法是结合两个运算符让 Jinja 翻译器正常工作，但是当我这样做时，我收到以下错误：ERROR - submit() takes from 3 to 4 positional arguments but 5 were given。

我对 Python 编码不是很熟悉，任何帮助都会很棒，请参阅下面我正在尝试构建的运算符的代码；

Python文件的头文件（请注意该文件包含本问题未提及的其他运算符）：

import ntpath
import os
import re
import time
import uuid
from datetime import timedelta

from airflow.contrib.hooks.gcp_dataproc_hook import DataProcHook
from airflow.contrib.hooks.gcs_hook import GoogleCloudStorageHook
from airflow.exceptions import AirflowException
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from airflow.version import version
from googleapiclient.errors import HttpError
from airflow.utils import timezone
from airflow.utils.operator_helpers import context_to_airflow_vars

修改后的DataprocHiveOperator：

class DataProcHiveOperator(BaseOperator):

template_fields = ['query', 'variables', 'job_name', 'cluster_name', 'dataproc_jars']
template_ext = ('.q',)
ui_color = '#0273d4'

@apply_defaults
def __init__(
        self,
        query=None,
        query_uri=None,
        hiveconfs=None,
        hiveconf_jinja_translate=False,
        variables=None,
        job_name='{{task.task_id}}_{{ds_nodash}}',
        cluster_name='cluster-1',
        dataproc_hive_properties=None,
        dataproc_hive_jars=None,
        gcp_conn_id='google_cloud_default',
        delegate_to=None,
        region='global',
        job_error_states=['ERROR'],
        *args,
        **kwargs):

    super(DataProcHiveOperator, self).__init__(*args, **kwargs)
    self.gcp_conn_id = gcp_conn_id
    self.delegate_to = delegate_to
    self.query = query
    self.query_uri = query_uri
    self.hiveconfs = hiveconfs or {}
    self.hiveconf_jinja_translate = hiveconf_jinja_translate
    self.variables = variables
    self.job_name = job_name
    self.cluster_name = cluster_name
    self.dataproc_properties = dataproc_hive_properties
    self.dataproc_jars = dataproc_hive_jars
    self.region = region
    self.job_error_states = job_error_states

def prepare_template(self):
    if self.hiveconf_jinja_translate:
        self.query_uri= re.sub(
            "(\$\{(hiveconf:)?([ a-zA-Z0-9_]*)\})", "{{ \g<3> }}", self.query_uri)

def execute(self, context):
    hook = DataProcHook(gcp_conn_id=self.gcp_conn_id,
                        delegate_to=self.delegate_to)

    job = hook.create_job_template(self.task_id, self.cluster_name, "hiveJob",
                                   self.dataproc_properties)

    if self.query is None:
        job.add_query_uri(self.query_uri)
    else:
        job.add_query(self.query)

    if self.hiveconf_jinja_translate:
        self.hiveconfs = context_to_airflow_vars(context)
    else:
        self.hiveconfs.update(context_to_airflow_vars(context))

    job.add_variables(self.variables)
    job.add_jar_file_uris(self.dataproc_jars)
    job.set_job_name(self.job_name)

    job_to_submit = job.build()
    self.dataproc_job_id = job_to_submit["job"]["reference"]["jobId"]

    hook.submit(hook.project_id, job_to_submit, self.region, self.job_error_states)

我希望能够在我的 HQL 代码中使用 Jinja 模板，以便在我的数据管道上实现分区自动化。

P.S：我将主要使用 Jinja 模板作为 Partition DateStamp

有人知道我收到的错误消息是什么+帮我解决吗？

ERROR - submit() takes from 3 to 4 positional arguments but 5 were given

谢谢！

Answer 1

这是因为第 5 个参数 job_error_states 仅在 master 中而不在当前稳定版本中 (1.10.1)。

1.10.1 的源代码 -> https://github.com/apache/incubator-airflow/blob/76a5fc4d2eb3c214ca25406f03b4a0c5d7250f71/airflow/contrib/hooks/gcp_dataproc_hook.py#L219

所以删除该参数，它应该可以工作。