我使用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
谢谢您!
1条答案
按热度按时间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
所以去掉这个参数,它就可以工作了。