如何使用python将jinja2传递给hiveql

disho6za  于 2021-06-27  发布在  Hive
关注(0)|答案(1)|浏览(382)

我使用gcloud composer作为我的气流。当我尝试在hql代码中使用jinja时,它不能正确地翻译它。我知道hiveoperator有一个jinja转换器,我已经习惯了,但是dataprochiveoperator没有。
我曾尝试将hiveconf直接用于我的hql文件,但在将这些值设置到我的分区(即。 INSERT INTO TABLE 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模板作为分区日期戳
有人知道我收到的错误信息是什么吗+帮我解决吗? ERROR - submit() takes from 3 to 4 positional arguments but 5 were given 谢谢您!

mutmk8jj

mutmk8jj1#

这是因为第五个论点 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
所以去掉这个参数,它就可以工作了。

相关问题