Rerun a task with different params

[iJanki-gr]

Hello, is it possible to rerun a failed task in a dag but with different params?

I've looked around and I don't think is possible in airflow. You can only rerun the full dag with different params but not rerun just one task in a past DagRun with different parameters.

Is not this something desirable? Any suggestion on how this could be achieved?

[potiuk]

Not possible. Not without modifying manually database and code of Airflow. You should rather run whatever code you need to run manually if you need to do it.

[iJanki-gr]

Would it be possible with a plugin? I'd like to get to know the code better anyway, this could be a good exercise.
Do you have any pointers for this on how it could be achieved?

[iJanki-gr]

I looked a bit at the code/db. The simplest/dirty way to do it would be to update the conf field in DagRun and then clear the task.
conf is a DagRun global field, not specific to a task.

It would be sufficient to be able to edit a DagRun conf in the UI or API.

I guess this feature request is not very popular?

[potiuk]

First time hear.

But FYI if you raise a PR and make such changes and do them in the way that will be generic and reusable by others and automaticallly covered by unit tests, feel free to do it. There are no "requests" for features here. There are > 2500 contributors to airflow. There are maybe "ideas" and if someone likes the idea, and implements it then it gets implemented. Or even PR gets raised without feature "idea".

So implementing something you want in a PR (we do not require to have issue for that) is the easiest way to get feature like that in.

[iJanki-gr]

I don't think I can invest in making this a generic change without having some design discussion or get some interest it risking to not even get it merged it in the end.
I will play around with the idea first then try to get some more discussion here or in Slack.

[fr33w1f1]

I'm late but for those who still find a way to work around. Using BranchPythonOperator will get tasks you want to run and skip the rest. This is an example
airflow==2.10.2

from datetime import datetime, timedelta
import time
from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.models.param import Param

FACT_TABLES = ['table1', 'table2', 'table3']

def process_table(**context):
    task_id = context['task_instance'].task_id 
    table_name = task_id.replace("process_", "") 
    params = context['params']
    time.sleep(5)
    print(f"Processing {table_name} with date: {params['process_date']}")

def select_tables(**context):
    dag_conf = context['dag_run'].conf
    if not dag_conf or not dag_conf.get('tables'):
        return [f'process_{table}' for table in FACT_TABLES] 
    return [f'process_{table}' for table in dag_conf.get('tables')]

default_args = {
    'start_date': datetime(2024, 1, 1),
}

with DAG(
    'selective_table_processing',
    default_args=default_args,
    schedule_interval=None,  # Manual trigger
    params={
        'process_date': Param(
            default=(datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d'),
            type='string'
        )
    }
) as dag:
    
    select_tables_task = BranchPythonOperator(
        task_id='select_tables',
        python_callable=select_tables,
        provide_context=True
    )
    
    tasks = []
    for table in FACT_TABLES:
        task = PythonOperator(
            task_id=f'process_{table}',
            python_callable=process_table,
            provide_context=True,
            dag=dag
        )
        tasks.append(task)

    select_tables_task >> tasks

On UI, click Trigger DAG and put this into Configuration JSON

{
    "process_date": "2025-02-18",
    "tables": ["table1", "table3"]
}

Click trigger, you see task process_table2 is skipped (pink) and the rest run with process_date