Воздушный поток sql_path не может читать файлы sql при передаче в качестве переменной шаблона Jinja - PullRequest
Новая
       5

Воздушный поток sql_path не может читать файлы sql при передаче в качестве переменной шаблона Jinja

0 голосов
/ 26 мая 2020

Я пытаюсь использовать переменную шаблона Jinja вместо Variable.get ('sql_path'), чтобы избежать попадания в DB при каждом сканировании файла dag

Исходный код 

import datetime
import os
from functools import partial
from datetime import timedelta
from airflow.models import DAG,Variable
from airflow.contrib.operators.snowflake_operator import SnowflakeOperator
from alerts.email_operator import dag_failure_email

SNOWFLAKE_CONN_ID = 'etl_conn'

tmpl_search_path = []
for subdir in ['business/', 'audit/', 'business/transform/']:
    tmpl_search_path.append(os.path.join(Variable.get('sql_path'), subdir))


def get_db_dag(
    *,
    dag_id,
    start_date,
    schedule_interval,
    max_taskrun,
    max_dagrun,
    proc_nm,
    load_sql
):

    default_args = {
        'owner': 'airflow',
        'start_date': start_date,
        'provide_context': True,
        'execution_timeout': timedelta(minutes=max_taskrun),
        'retries': 0,
        'retry_delay': timedelta(minutes=3),
        'retry_exponential_backoff': True,
        'email_on_retry': False,
    }


    dag = DAG(
        dag_id=dag_id,
        schedule_interval=schedule_interval,
        dagrun_timeout=timedelta(hours=max_dagrun),
        template_searchpath=tmpl_search_path,
        default_args=default_args,
        max_active_runs=1,
        catchup='{{var.value.dag_catchup}}',
        on_failure_callback=alert_email_callback,
    )


    load_table = SnowflakeOperator(
        task_id='load_table',
        sql=load_sql,
        snowflake_conn_id=SNOWFLAKE_CONN_ID,
        autocommit=True,
        dag=dag,
    )

    load_vcc_svc_recon

    return dag

# ======== DAG DEFINITIONS #

edw_table_A = get_db_dag(
    dag_id='edw_table_A',
    start_date=datetime.datetime(2020, 5, 21),
    schedule_interval='0 5 * * *',
    max_taskrun=3,  # Minutes
    max_dagrun=1,  # Hours
    load_sql='recon/extract.sql',
)

Когда я заменил Variable.get('sql_path') шаблоном Jinja '{{var.value.sql_path}}', как показано ниже, и запустил Dag, он выдал ошибку, как показано ниже, он не смог получить файл из подкаталога папки SQL 

tmpl_search_path = []
for subdir in ['bus/', 'audit/', 'business/snflk/']:
    tmpl_search_path.append(os.path.join('{{var.value.sql_path}}', subdir))

Получена ошибка ниже как inja2.exceptions.TemplateNotFound: extract. sql

1 Ответ

1 голос
/ 26 мая 2020

Шаблоны не отображаются везде в сценарии DAG. Обычно они отображаются в шаблонных параметрах операторов. Таким образом, если вы не передадите элементы tmpl_search_path в какой-то шаблонный параметр, {{var.value.sql_path}} не будет отображаться.

template_searchpath из DAG не является шаблоном. Вот почему вы не можете передавать ему шаблоны Jinja.

Возможные варианты:

  1. Жестко закодируйте значение Variable.get('sql_path') в конвейерном скрипте.
  2. Сохраните значение Variable.get('sql_path') в файле конфигурации и прочтите его оттуда в скрипте конвейера.
  3. Переместите вызов Variable.get() из for-l oop. В результате будет в три раза меньше запросов к базе данных.

Подробнее о шаблонах в Airflow.

...