https://prefect.io logo
Powered by
Title
c

Chu

07/24/2022, 7:39 PM
hi Community, I have 2 flows: A and B, their dependency is 
A -> B
, both A and B will take a list of same org_ids as parameters (need to use map function for parallel running) when I use a parent flow to schedule flow A and B, can I use map function to pass parameters like this way? or is there a better way to do that? code in the thread
✅ 1
a

Anna Geller

07/24/2022, 8:38 PM
can you move code block to the thread?
✅ 1
this is the syntax you are looking for https://discourse.prefect.io/t/how-to-map-over-flows-with-various-parameter-values/365
c

Chu

07/24/2022, 8:42 PM
with Flow (name="parent flow", schedule = CronSchedule("0 0 * * *"), executor=LocalDaskExecutor() ) as flow:

  A_flow_id = create_flow_run(
                                     flow_name="A",
                                     project_name="sample project", 
                                     task_args= ...
                                                  )

  wait_for_A_flow = wait_for_flow_run(
        flow_run_id = A_flow_id
    )

  B_flow_id = create_flow_run(
                                                        flow_name = "B",  
                                            project_name = "sample project",
                                            upstream_tasks = [wait_for_A_flow], 
                                            task_args= ...
                                                  )

  wait_for_B_flow = wait_for_flow_run(
        flow_run_id = B_flow_id
    )

flow.run_config = KubernetesRun()
🙌 1
@Anna Geller I altered the scripts above, I think that should work? Following that structure, I’m wondering if my flow A and B is used to run dbt jobs, for example can flow A be defined as this? I’m a bit confusing 1. how can I pass the params from parent flow to child flows? 2. do I need to define those params again in child flow? 3. and if child flow still need map function? 
with Flow(name="A") as flow:

  a_list_of_orgs_params = Parameter("a_list_of_orgs")
  start_date_param = Parameter("start_date", default="2012-01-01 +0000")
  end_date_param = Parameter("end_date", default="2012-01-03 +0000")
        
    dbt_run.map(       
        org_id=a_list_of_orgs_params,
        start_date=unmapped(start_date_param),
        end_date=unmapped(end_date_param),
        upstream_tasks = unmapped([some_task])
     )
a

Anna Geller

07/25/2022, 3:29 PM
the same way as with any task
do I need to define those params again in child flow? • if it's required parameter and you didn't set a default value, then yes
#3 you can do iterated maping with create_flow_run task as with any other task and the result will be a list of flow run IDs
just posted a couple of more examples https://discourse.prefect.io/t/how-to-map-over-flows-with-various-parameter-values/365/2
2 Views
#prefect-communityJoin Slack
Powered by