Airflow DAG Dépendance des tâches dans une boucle

Question

J'ai un DAG qui doit recompiler ces listes de clients dans différentes marques. Le script est appelé avec deux arguments brand et listtype.

J'ai besoin que les marques s'exécutent simultanément, mais que les types de liste dépendent du type de liste précédent, mais je n'arrive pas à trouver comment faire cela dans une boucle. Pouvez-vous m'aider ?

BrandsToRun = ['A', 'B', 'C']
ListTypes = ['1', '2', '3']

# Defining the DAG
################################################################################
with DAG(
        'MusterMaster',
        default_args = default_args,
        description = 'x',        
        # schedule_interval = None
        schedule_interval = '30 4 * * *',
        catchup = False
        ) as MusterMaster:

        for Brand in BrandsToRun:
            for ListType in ListTypes:

                ListLoad = BashOperator(
                                        task_id='Load_'+str(Brand)+'_'+str(ListType),
                                        bash_command = """python3 '/usr/local/bin/MusterMaster.py' {0} {1}""".format(Brand[0], ListType[0]),
                                        pool='logs'
                                        )

ListLoad

Je veux que les tâches aient une structure de dépendance comme celle-ci, mais je n'arrive pas à la comprendre. Brand devrait s'exécuter simultanément, mais les ListTypes devraient dépendre du ListType précédent.

Muster A 1 >> Muster A 2 >> Muster A 3

Muster B 1 >> Muster B 2 >> Muster B 3

Muster C 1 >> Muster C 2 >> Muster C 3

Quelle est la meilleure façon d'y parvenir ?

Answer 1

Vous pouvez le faire :

    for Brand in BrandsToRun:
        list = []
        for ListType in ListTypes:
            list.append(BashOperator(
                task_id='Load_'+str(Brand)+'_'+str(ListType),
                bash_command = """python3 '/usr/local/bin/MusterMaster.py' {0} {1}""".format(Brand[0], ListType[0]),
                pool='logs'))
            if len(list) > 1:
                list[-2] >> list[-1]

Ce qui vous donnera :