如何让 apache 气流用 Jinja 渲染 Hive HQL ${variables}答案

【问题标题】：How to get apache airflow to render Hive HQL ${variables} with Jinja如何让 apache 气流用 Jinja 渲染 Hive HQL ${variables}
【发布时间】：2017-05-04 05:40:56
【问题描述】：

似乎支持这一点，您传入带有 ${xxx} 变量的 HQL 脚本，并在实际执行模板渲染然后替换的阶段之前对其进行预处理以将它们转换为 {{xxx}} Jinja 样式那些具有来自用户提供的字典的值。我相信这是因为 HiveOperator 类中有这样的功能：

def prepare_template(self):
    if self.hiveconf_jinja_translate:
        self.hql = re.sub(
            "(\$\{([ a-zA-Z0-9_]*)\})", "{{ \g<2> }}", self.hql)
    if self.script_begin_tag and self.script_begin_tag in self.hql:
        self.hql = "\n".join(self.hql.split(self.script_begin_tag)[1:])

问题是我不知道如何在模板渲染阶段之前触发这段代码被调用。我有一个像这样的基本 DAG 脚本：

from airflow import DAG
from airflow.operators.hive_operator import HiveOperator
from datetime import datetime, timedelta

default_args = dict(
    owner='mpetronic',
    depends_on_past=False,
    start_date=datetime(2017, 5, 2),
    verbose=True,
    retries=1,
    retry_delay=timedelta(minutes=5)
    )

dag = DAG(
    dag_id='report',
    schedule_interval='* * * * *',
    user_defined_macros=dict(a=1, b=2),
    default_args=default_args)

hql = open('/home/mpetronic/repos/airflow/resources/hql/report.hql').read()

task = HiveOperator(
    task_id='report_builder',
    hive_cli_conn_id='hive_dv',
    schema='default',
    mapred_job_name='report_builder',
    hiveconf_jinja_translate=True,
    dag=dag,
    hql=hql)

我可以看到我的 user_defined_macros 字典在代码中与全局 jinja 上下文字典合并，然后将其应用于我的 HQL 脚本以将其呈现为模板。但是，因为我的 HQL 是原生 HQL，所以我想要更新的所有变量都是 ${xxx} 的形式，而 jinja 只是跳过它们。我需要气流首先调用 prepare_template() 但只是不知道如何实现。

我意识到我可以手动将我的 HQL ${xxx} 更改为 {{xxx}} ，但这似乎是一种反模式。我希望脚本能够在本地或通过气流工作。这是函数，在 TaskInstance 类中，确实呈现了我手动更改的 {{xxx}} 值：

def render_templates(self):
    task = self.task
    jinja_context = self.get_template_context()
    if hasattr(self, 'task') and hasattr(self.task, 'dag'):
        if self.task.dag.user_defined_macros:
            jinja_context.update(
                self.task.dag.user_defined_macros)

    rt = self.task.render_template  # shortcut to method
    for attr in task.__class__.template_fields:
        content = getattr(task, attr)
        if content:
            rendered_content = rt(attr, content, jinja_context)
            setattr(task, attr, rendered_content)

【问题讨论】：

能够在气流 1.8 中正常工作，hql 文件中的 ${xxx} 表单变量将替换为 user_defined_macros 中的相应条目。您是否尝试在代码的上述阶段添加记录器？就我而言，我能够看到相应替换的调用。

标签： hive jinja2 airflow

【解决方案1】：

我发现了我的问题。这是上述方法中使用的正则表达式：

(\$\{([ a-zA-Z0-9_]*)\})

它不考虑直线变量的形式：

${hivevar:var_name}

它不考虑模式中的冒号。这种形式是使用 beeline 在命名空间内定义 Hive 变量的更标准方法。要使这个 Jinja 替换工作，您必须使用 ${var_name} 引用 HQL 中的变量，但您只能使用以下方式在直线中定义变量：

set hivevar:var_name=123;

我认为当你使用 beeline 运行时，Airflow 应该完全支持命名空间变量的 hivevar:var_name 样式，因为 beeline 是与 Hive 一起使用的首选客户端。

【讨论】：