title: How to pass a Jinja2 to HiveQL using Python
body: <p>I'm using Gcloud Composer as my Airflow. When I try to use Jinja in my HQL code, it does not translate it correctly.
  I know that the <a href="https://github.com/apache/incubator-airflow/blob/master/airflow/operators/hive_operator.py" rel="nofollow noreferrer">HiveOperator</a> has a Jinja translator as I'm used to it, but the <a href="https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/operators/dataproc_operator.py" rel="nofollow noreferrer">DataProcHiveOperator</a> doesn't.</p>
  
  <p>I've tried to use the HiveConf directly into my HQL files, but when setting those values to my Partition (i.e. <code>INSERT INTO TABLE abc PARTITION (ds = ${hiveconf:ds})</code>)`, it doesn't work. 
  I have also added the following to my HQL file:</p>
  
  <pre><code>SET ds=to_date(current_timestamp());
  
  SET hive.exec.dynamic.partition=true;
  SET hive.exec.dynamic.partition.mode=nonstrict;
  </code></pre>
  
  <p>But it didn't work as HIVE is transforming the formula above into a STRING.</p>
  
  <p>So my idea was to combine both operators to have the Jinja translator working fine, but when I do that, I get the following error: <code>ERROR - submit() takes from 3 to 4 positional arguments but 5 were given</code>.</p>
  
  <p>I'm not very familiar with Python coding and any help would be great, see below code for the operator I'm trying to build;</p>
  
  <p>Header of the Python File (please note that the file contains other Operators not mentioned in this question):</p>
  
  <pre><code>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
  </code></pre>
  
  <p>modified DataprocHiveOperator:</p>
  
  <pre><code>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&lt;3&gt; }}", 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)
  </code></pre>
  
  <p>I would like to be able to use Jinja templating inside my HQL code to allow partition automation on my data pipeline. </p>
  
  <p>P.S: I'll use the Jinja templating mostly for Partition DateStamp</p>
  
  <p>Does anyone know what is the error message I'm getting + help me solve it?</p>
  
  <p><code>ERROR - submit() takes from 3 to 4 positional arguments but 5 were given</code></p>
  
  <p>Thank you!</p>