Параметр воздушного потока BashOperator от значения XCom - PullRequest
Новая
       19

Параметр воздушного потока BashOperator от значения XCom

0 голосов
/ 26 апреля 2020

У меня возникли проблемы с назначением значения xcom для BashOperator.

Все параметры извлекаются правильно, кроме tmp_dir, который является значением xcom, сгенерированным во время init_dag. Я смог получить значение в своем пользовательском операторе, но не смог сделать это в BashOperator. Я добавил результаты трех разных способов, которые я попробовал, которые пришли мне в голову.

Я думаю, что одним из способов могло бы быть, если бы я мог сохранить это значение в переменной, но я также не смог понять, как это сделать.

Любая помощь будет высоко оценена.

Вот мой код DAG: 

import airflow
from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.models import Variable
from utility import util
import os

from airflow.operators.bash_operator import BashOperator

from operators.mmm_operator import MMMOperator  #it is a custom operator
from operators.iftp_operator import IFTPOperator #it is another custom operator

AF_DATAMONTH = util.get_date_by_format(deltaMth=2,deltaDay=0,ft='%b_%Y').lower() #it gives a date in required format
AF_FILENM_1 = 'SOME_FILE_' + AF_DATAMONTH + '.zip'  #required filename for ftp

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(0),
 }

dag = DAG(dag_id='my_dag', default_args=default_args, schedule_interval=None)


init_dag = MMMOperator(
    task_id='init_dag',
    provide_context=True,
    mmm_oracle_conn_id=Variable.get('SOME_VARIABLE'),
    mmm_view="{0}.{1}".format(Variable.get('ANOTHER_VARIABLE'), AF_DAG_MMM_VIEW_NM),
    mmm_view_filter=None,
    mmm_kv_type=True,
    mmm_af_env_view="{0}.{1}".format(Variable.get('ANOTHER_VARIABLE_1'),Variable.get('ANOTHER_VARIABLE_2')),
    dag=dag
)  #local_tmp_folder is generated here and pushed via xcom


download_ftp_files = IFTPOperator(task_id='download_ftp_files',
                    ftp_conn_id=util.getFromConfig("nt_conn_id"), #value properly retrieved by xcom_pull  
                    operation='GET',
                    source_path=util.getFromConfig("nt_remote_folder"), #value properly retrieved by xcom_pull
                    dest_path=util.getFromConfig("local_tmp_folder"), #value properly retrieved by xcom_pull
                    filenames=AF_FILENM,                    
                    dag=dag
)



bash_cmd_template = "cd /vagrant/ && python3 hello_print.py {{params.client}} {{params.task}} {{params.environment}} {{params.tmp_dir}} {{params.af_file_nm}}"

#try 1 output value for params.tmp_dir: {{ ti.xcom_pull(task_ids="init_dag")["local_tmp_folder"] }} - instead of the actual tmp folder location
#try 2 and try 3 output: Broken DAG: [/home/vagrant/airflow/dags/my_dag.py] name 'ti' is not defined - message in UI

execute_main_py_script = BashOperator(
    task_id='execute_main_py_script',
    bash_command=bash_cmd_template,
    params={'client' : 'some_client',
            'task' : 'load_some_task',
            'environment' : 'environment_name',
            #'tmp_dir' : util.getFromConfig("local_tmp_folder"),   #try 1
            #'tmp_dir' : {{ ti.xcom_pull(task_ids="init_dag")["local_tmp_folder"] }} #try 2
            #'tmp_dir' : ti.xcom_pull(task_ids="init_dag")["local_tmp_folder"] #try 3
             'af_file_nm' : AF_FILENM_1
            },
    provide_context=True,        
    dag=dag
)


init_dag >> download_ftp_files >> execute_main_py_script

1 Ответ

1 голос
/ 27 апреля 2020

Аргумент params в BashOperator не является шаблоном Jinja, поэтому любые значения, передаваемые вами в params, будут отображаться "как есть".

Вы должны напрямую передать значение tmp_dir в bash_cmd_template следующим образом: 

bash_cmd_template = """
cd /vagrant/ && python3 hello_print.py {{params.client}} {{params.task}} {{params.environment}} {{ ti.xcom_pull(task_ids="init_dag")["local_tmp_folder"] }} {{params.af_file_nm}}
"""


execute_main_py_script = BashOperator(
    task_id='execute_main_py_script',
    bash_command=bash_cmd_template,
    params={'client' : 'some_client',
            'task' : 'load_some_task',
            'environment' : 'environment_name',
            'af_file_nm' : AF_FILENM_1
            },
    provide_context=True,        
    dag=dag
)
...