Sequential Execution of Dynamic Mapped Task

[dimitar-petrov]

Hello, is it possible to define a sequential dependencies for dynamically mapped tasks.

The case is as follows

list_s3_files -> extract -> transform -> load

extract, transform and load steps are expanded for all files from first task.

Since files are partitioned by timestamp field the load step should add to incremental dataset following the order of files generated by list_s3_files.

Thanks

[potiuk]

No. dynamic tasks are run in parallel and that's the only reason why you would like to use them this way. If you want to process your file sequentially, it makes completely no sense to make it Dynamic Tasks. you can write your custom (non-dynamic) task:

@task
def my_processing(): 
    for file in files:
        process(file)

If you want to, however do some processing in parallel and THEN join the result of it sequentially (typical map-reduce scenario) then you can run the dynamic tasks for each file separately (and store it somewhere - like s3) and THEN run a single task that will take the result of those paralll-processed files and join them together. Dynamic Task mapping has full support for it - it can pas the "array" of outputs (for example output files produced) and your next task can take the array and process all files from the arrray as it finds appropriately.

[dimitar-petrov]

Alright, I got the point:

I went with alternative approach and decided to create the tasks manually and define the dependencies .

Here is an example DAG.

# Scratch buffer for: python-mode

default_args = {
    "start_date": datetime(2022, 8, 24),
    "owner": "owner",
    "depends_on_past": False,
    "retries": 5,
    "params": {
        "categories": ["cat1", "cat2"],
    },
}


def etl_step(operation: str, **kwargs):
    arguments = build_args_list(**kwargs)

    return KubernetesPodOperator(
        namespace="prod",
        name="source_data_etl",
        task_id=f"{operation}_"
        + "_".join([f"{k}_{v}" for k, v in kwargs.items() if v is not None]),
        image="registry.example.com/etl_img:{tag}".format(tag="latest"),
        arguments=[operation] + arguments,
        env_vars={"PYTHONUNBUFFERED": "1",},
        secrets=env_var_secrets,
        is_delete_operator_pod=True,
        in_cluster=True,
        get_logs=True,
        startup_timeout_seconds=900,
        node_selector={"accelerator": "vm-cortex"},
    )


def etl_pipeline(categories **kwargs):
    for category in categories:
        groups = []
        for cli_args in get_source_data_objects(category):
            with TaskGroup(group_id="{category}_{date}".format(**cli_args)) as tg:
                etl_pipeline = [
                    {"operation": "ExtractSourceData", **cli_args},
                    # {"operation": "ValidateSourceData", **cli_args, "env": "raw"},
                    {"operation": "TransformSourceData", **cli_args},
                    {"operation": "ValidateSourceData", **cli_args, "env": "stage"},
                    {"operation": "LoadSourceData", **cli_args},
                ]

                tasks = [etl_step(**step_kws) for step_kws in etl_pipeline]

                for index in range(len(tasks) - 1):
                    tasks[index] >> tasks[index + 1]

                if groups:
                    tasks[-1] << groups[-1]

                groups.append(tg)

with DAG(
    "binance_cmfutures_trades",
    schedule_interval=None,
    default_args=default_args,
    concurrency=1,
) as dag:
    etl_pipeline(**default_args["params"])

That DAG works as expected but I would like to be able to parametrize the categories when running manually.

Is there a way to access params outside of a task, in a regular python function to be able to loop over the categories and create the tasks groups needed.

Thanks in advance.

[potiuk]

You need to create DAG "object" during parsing the DAG python file. This is how Airllow works. And you can do whatever you want as long as you follow best practices https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html?highlight=best+practices - just follow it. I am not sure what you mean by " in a regular python function to be able to loop over the categories and create the tasks groups needed. ". The best practices document desrcribe all the various ways you can create dynamic dags - read it and see if gives you answers.

[manugarri]

@potiuk here is a use case that I believe merits the option to expand the output of a task but require sequential order.

I am parsing a yaml with some steps to run as a @task , and I cannot add a task inside with a for loop to create the actual operators that run the operations described on the yaml file.

The dag looks more or less like this :

with DAG(...):
  @task
  def parse_yaml(yaml_path):
     """ 
     read the templated yaml and expand the jinja variables inside
     This yaml has blocks of steps that have to run sequentially:
      something like this:
      groups:
        - group_name:1
          steps:
            - step_name: 1.1
              step_emr_command: .....
        - group_name:2
          steps:
             - step_name: 2.1
               step_emr_command: ...
      """
     return parsed_yaml
  
  @task_group
  def run_group(group_definition):
     setup_emr_cluster = EMRCreateJobFlowOperator()
     run_steps = EMRAddStepsOperator(group_definition['steps'])
     terminate_emr_cluster = EMRTerminateJobFlowOperator()
     setup_emr_cluster >> run_steps >> terminate_emr_cluster
  
  parsed_yaml = parse_yaml(yaml_path)
  
  # this runs all the step groups in parallel (bad)
  run_group.expand(parsed_yaml)

hopefully this pseudo example showcases what im trying to achieve. I need the run_group for group:1 to run before group2 , not in parallel.

Maybe there is a way to do this natively on airflow, I have tried this:

with DAG(...):
  @task
  def parse_yaml(yaml_path):
     """ 
     read the templated yaml and expand the jinja variables inside
     This yaml has blocks of steps that have to run sequentially:
      something like this:
      groups:
        - group_name:1
          steps:
            - step_name: 1.1
              step_emr_command: .....
        - group_name:2
          steps:
             - step_name: 2.1
               step_emr_command: ...
      """
     return parsed_yaml
  
  @task
  def run_group_task(parsed_yaml):
     for group parsed_yaml:
       create_cluster = EMRCreateJobFlowOperator()
       run_steps = EMRAddStepsOperator(group_definition['steps'])
       terminate_emr_cluster = EMRTerminateJobFlowOperator()
       setup_emr_cluster >> run_steps >> terminate_emr_cluster
  
  parsed_yaml = parse_yaml(yaml_path)
  
  # this runs all the step groups in parallel (bad)
  run_group_task(parsed_yaml)

The above dag throws an exception inside the run_group_task:

airflow.exceptions.AirflowException: Tried to create relationships between tasks that don't have DAGs yet. Set the DAG for at least one task and try again: [<Task(EmrAddStepsOperator): run_steps>, <Task(EmrCreateJobFlowOperator): create_cluster>]

[wolfgesbro]

No. dynamic tasks are run in parallel and that's the only reason why you would like to use them this way. If you want to process your file sequentially, it makes completely no sense to make it Dynamic Tasks.

I do not agree - dynamic tasks make sense to me also with sequential processing: They appear on the UI, and you can inspect their status without reading logs.

Thanks to the introduction of the max_active_tis_per_dagrun parameter you can now easily limit the dynamic tasks to sequential scheduling (by setting that parameter to 1).