airflow access dag_run.conf values and pass it to SSHOperator command

Question

If I use this json parameter to trigger a DAG {"number":["1","2","3"], "items":["a", "b", "c"]}, how can I use dag_run.conf to access the values and pass them to the ssh command in the SSHoperator task?

For example

def run_config(**kwargs):
    number_input = kwargs['dag_run'].conf.get('number')
    items_input = kwargs['dag_run'].conf.get('items')
    return number_input, items_input

#assign the returned values from the run_config function to the following variables
number, item = run_config

#use the above 2 variables as argument in the following ssh command call 
with DAG(blah blah blah) as dag:

    for i in range(0, len(number)):
        process_task = SSHOperator(
                    task_id='run_'.format(number[i]),
                    command='run_somthing.sh' + ' ' + number[i] + ' ' + item[i]
                    )

Hussein Awala · Accepted Answer · 2022-08-11T23:05:38.807

0

If you are using a version >= 2.3.0, you can use the Dynamic Task Mapping:

from airflow import DAG
from airflow.decorators import task
from airflow.providers.ssh.operators.ssh import SSHOperator

with DAG(dag_id="...") as dag:
    @task
    def get_config(**context):
        return context['dag_run'].conf.get('items')

    items_to_process = get_config()

    SSHOperator.partial(
        task_id='run_ssh'
    ).expand(
        command=[f"run_something.sh {ind} {val}" for ind, val in enumerate(items_to_process)]
    )

For the versions < 2.3.0, you can inspire from this answer

edited Aug 11 '22 at 23:05

answered Aug 11 '22 at 17:26

Hussein Awala

4,285
2
9
23

hi, I just updated my original question. not sure if your answer still applies. – Dozel Aug 11 '22 at 17:59
yes it still applies, where you call `number, item = run_config` outside the run context, where you should run them in a task and pass the result to the second task by `xCom` or airflow variables – Hussein Awala Aug 11 '22 at 18:02
but do I have to set those 2 variables as airflow variables (using variable.set)? Can I just use python variables instead? – Dozel Aug 11 '22 at 18:08
I just updated my answer, can you check it? – Hussein Awala Aug 11 '22 at 23:06
Thank you Hussein. I will check it out at work tomorrow. – Dozel Aug 11 '22 at 23:47

score 0 · Answer 2 · answered Aug 17 '22 at 22:32

actually, I could use the jinja template directly to get the trigger parameter value into any operator without using a function nor pythonOperator to call it.

will mark the answer from @hussein Awala as correct since a dynamic creating tasks method is introduced, for which can't be done with my solution below.

process_task = SSHOperator(
        task_id='run_job',
        command=shell_command + ' ' + "{{ dag_run.conf['number'] }}" + ' ' + "{{ dag_run.conf['item'] }}"
)

airflow access dag_run.conf values and pass it to SSHOperator command

2 Answers2