Airflow best way to skipping task?

Question

I'm trying to run tasks independently and in parallel,

I've dag look like this:

                                      ---> patternA   ---> file1a
                                                      ---> file2a
                                                      ---> file3a
sensor ---> move_csv ---> result_mv                                   ---> rerun_dag

                                      ---> patternB   ---> file1b
                                                      ---> file2b
                                                      ---> file3b

my dag.py:

sensor = FileSensor(
    task_id="sensor ",
    filepath=filePath,
    fs_conn_id='airflow_db',
    poke_interval=10,
    dag=dag,
)
move_csv = BranchPythonOperator(
    task_id='move_csv',
    python_callable=moveCsvFile,
    trigger_rule='none_failed',
    dag=dag,
)
result_mv = BranchPythonOperator(
    task_id='result_mv',
    python_callable=result,
    trigger_rule='none_failed',
    dag=dag,
)
pattern_A = DummyOperator(
    task_id="pattern_A ",
    dag=dag,
)
pattern_B = DummyOperator(
    task_id="pattern_B ",
    dag=dag,
)
file1 = BashOperator(
    task_id="file1a ",
    bash_command='python3 '+scriptPath+'file1.py "{{ execution_date }}"',
    trigger_rule='one_success',
    dag=dag,
)
file2 = BashOperator(
    task_id="file2a",
    bash_command='python3 '+scriptPath+'file2.py "{{ execution_date }}"',
    trigger_rule='one_success',
    dag=dag,
)
file3 = BashOperator(
    task_id="file3a",
    bash_command='python3 '+scriptPath+'file3.py "{{ execution_date }}"',
    trigger_rule='one_success',
    dag=dag,
)
file1 = BashOperator(
    task_id="file1b ",
    bash_command='python3 '+scriptPath+'file1b.py "{{ execution_date }}"',
    trigger_rule='one_success',
    dag=dag,
)
file2 = BashOperator(
    task_id="file2b",
    bash_command='python3 '+scriptPath+'file2b.py "{{ execution_date }}"',
    trigger_rule='one_success',
    dag=dag,
)
file3 = BashOperator(
    task_id="file3b",
    bash_command='python3 '+scriptPath+'file3b.py "{{ execution_date }}"',
    trigger_rule='one_success',
    dag=dag,
)
move_csv.set_upstream(sensor)
result_mv.set_upstream(move_csv)
patternA.set_upstream(result_mv)
patternB.set_upstream(result_mv)
file1a.set_upstream(patternA)
file2a.set_upstream(patternA)
file3a.set_upstream(patternA)
file1b.set_upstream(patternB)
file2b.set_upstream(patternB)
file3b.set_upstream(patternB)
rerun.set_uptstream( from all file ...)

what is the best way, in patternA to skip file2a and file3a if i only have file1a matching pattern ? And if i've file1a and file2a matching i'd like to run them in parallel and skip file3a.

My files task are running a python script call with a BashOperator.

Thanks for help ! :)

Airflow best way to skipping task?

Answers (1)

Related Questions