我使用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 谢谢您！