Hi Flow A repeatedly loop run Flow B I want run next instanc

Question

Hi Flow A repeatedly loop run Flow B I want run next instance of Flow B after current Flow B is completed I ran below and 2 instances of the same flow run at same time and flow id is same see screenshot ```with Flow parent flow as flow for i in range 2 flow a = create flow run flow name= livelots flow project name= LiveLots ETL wait for flow a = wait for flow run flow a raise final state=True print i ```

Anna Geller · Answer

A couple of issues here 1 You can t run for loops in Prefect 1 0 This is possible in Prefect 2 0 but in 1 0 within the Flow you can only construct your DAG i e call your tasks no for loops allowed Technically your for loop works because it iterates over a static list of values but still this is not something you should do in 1 0 mapping would be a better approach Check the example below 2 The print statement is also not allowed ```from prefect import Flow unmapped from prefect tasks prefect import create flow run from prefect executors import LocalDaskExecutor with Flow parent flow executor=LocalDaskExecutor as parent flow mapped flow run ids = create flow run map flow name= flow name 1 flow name 2 flow name 3 project name=unmapped your project name ```

Anna Geller · Answer

And actually regarding your issue it s most likely due to the idempotency key If you add `idempotency key=datetime now strftime m d Y H M S ` this should work ```with Flow parent flow as flow for i in range 2 flow a = create flow run flow name= livelots flow project name= LiveLots ETL idempotency key=datetime now strftime m d Y H M S wait for flow a = wait for flow run flow a raise final state=True ```

Patrick Tan · Answer

I added the code and now child flow only execute once I expect it to run twice for i in range 2

Patrick Tan · Answer

< Anna Geller> miss your 1st response let me review first

Kevin Kho · Answer

Yeah this looks like an idempotency key issue The docstring of the idempotency key in the task gives a good description of what is happening

Patrick Tan · Answer

< Kevin Kho> I have a parent flow with a task that return a list of files then for each file I want to run a child flow by passing filename as parameter I am scratching my head since task is not iterable do you any idea or sample code

Kevin Kho · Answer

You can use ```create flow run map ``` and it s analogous to a for loop You just need to supply unique idempotency keys

Anna Geller · Answer

this is without subflows but a very similar use case to yours <https discourse prefect io t how to sequentially loop several cleanup tasks over a set of files to process for etl 370>

Anna Geller · Answer

```from prefect import task Flow Parameter from prefect executors import LocalExecutor from datetime import timedelta task max retries=5 retry delay=timedelta minutes=10 def read data from file file name str return some data task max retries=5 retry delay=timedelta minutes=10 def etl 1 x some cleaning return x task max retries=5 retry delay=timedelta minutes=10 def etl 2 x some cleaning return x task max retries=5 retry delay=timedelta minutes=10 def etl 15 x some cleaning return x with Flow clean all files executor=LocalExecutor as flow all files = Parameter files default= file1 csv file2 csv file100 csv data = read data from file map all files transformed data = etl 1 map data transformed data = etl 2 map transformed data transformed data = etl 15 map transformed data ``` you could basically do the same but instead of `etl 1 map data ` it would be `create flow run map args `

Patrick Tan · Answer

< Anna Geller> Can you tell me what s wrong with my code I did create flow run map for all the files which will be passed as parameters then create flow run Sorry I am new on this with Flow parent flow as flow mapped flow run ids = create flow run map parameters= config file Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml bucket wp ahdata sandbox prefix dot scrapy live lots grays test live lots raw tsv config file Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml bucket wp ahdata sandbox prefix dot scrapy live lots Blackwell 20220325 180155 Blackwell DS=420 tsv flow a = create flow run flow id=mapped flow run ids flow name= livelots flow project name= LiveLots ETL idempotency key=datetime now strftime m d Y H M S wait for flow a = wait for flow run flow a raise final state=True flow run

Kevin Kho · Answer

What is the error here Still only one waits created

Patrick Tan · Answer

This is the error 2022 04 01 15 50 49 0400 INFO prefect FlowRunner | Beginning Flow run for parent flow 2022 04 01 15 50 49 0400 INFO prefect TaskRunner | Task create flow run Starting task run 2022 04 01 15 50 49 0400 INFO prefect TaskRunner | Task create flow run Finished task run for task with final state Mapped 2022 04 01 15 50 49 0400 INFO prefect TaskRunner | Task create flow run 0 Starting task run 2022 04 01 15 50 49 0400 ERROR prefect TaskRunner | Task create flow run 0 Exception encountered during task execution Traceback most recent call last File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect engine task runner py line 880 in get task run state logger=self logger File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect tasks prefect flow run py line 119 in create flow run Both `flow id` and `flow name` are null You must pass a flow ValueError Both `flow id` and `flow name` are null You must pass a flow identifier 2022 04 01 15 50 49 0400 INFO prefect TaskRunner | Task create flow run 0 Finished task run for task with final state Failed 2022 04 01 15 50 49 0400 INFO prefect TaskRunner | Task create flow run 1 Starting task run 2022 04 01 15 50 49 0400 ERROR prefect TaskRunner | Task create flow run 1 Exception encountered during task execution Traceback most recent call last File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect engine task runner py line 880 in get task run state logger=self logger File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect tasks prefect flow run py line 119 in create flow run Both `flow id` and `flow name` are null You must pass a flow ValueError Both `flow id` and `flow name` are null You must pass a flow identifier 2022 04 01 15 50 49 0400 INFO prefect TaskRunner | Task create flow run 1 Finished task run for task with final state Failed 2022 04 01 15 50 49 0400 INFO prefect TaskRunner | Task create flow run Starting task run 2022 04 01 15 50 49 0400 INFO prefect TaskRunner | Task create flow run Finished task run for task with final state TriggerFailed 2022 04 01 15 50 50 0400 INFO prefect TaskRunner | Task wait for flow run Starting task run 2022 04 01 15 50 50 0400 INFO prefect TaskRunner | Task wait for flow run Finished task run for task with final state TriggerFailed 2022 04 01 15 50 50 0400 INFO prefect FlowRunner | Flow run FAILED some reference tasks faile

Patrick Tan · Answer

When running child flow flow name and flow project are same only parameters are different So I only map parameters ```mapped flow run ids = create flow run map parameters= config file Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml bucket wp ahdata sandbox prefix dot scrapy live lots grays test live lots raw tsv config file Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml bucket wp ahdata sandbox prefix dot scrapy live lots Blackwell 20220325 180155 Blackwell DS=420 tsv ```

Kevin Kho · Answer

`create flow run` needs either the flow id or flow name and project name I don t think you supply it

Patrick Tan · Answer

I provides with flow name flow project and parameter parameter is list of 2 dictionaries Then I follow by create flow run with flow id from and idempotency key The error shows failure to run child flow twice ``` with Flow parent flow as flow filelist = get filelist task mapped flow run ids = create flow run map flow name= livelots flow project name= LiveLots ETL parameters= config file Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml bucket wp ahdata sandbox prefix dot scrapy live lots grays test live lots raw tsv config file Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml bucket wp ahdata sandbox prefix dot scrapy live lots Blackwell 20220325 180155 Blackwell DS=420 tsv flow a = create flow run flow id=mapped flow run ids idempotency key=datetime now strftime m d Y H M S wait for flow a = wait for flow run flow a raise final state=True flow run ``` Users patricktan bitbucket data live lots etl venv livelotsetl bin python Users patricktan bitbucket data live lots etl live lots etl batch flows schedule flow py Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml wp ahdata sandbox dot scrapy live lots Blackwell 20220325 180155 Blackwell DS=420 tsv 2022 04 01 16 16 34 0400 INFO prefect FlowRunner | Beginning Flow run for parent flow 2022 04 01 16 16 34 0400 INFO prefect TaskRunner | Task create flow run Starting task run 2022 04 01 16 16 34 0400 INFO prefect TaskRunner | Task create flow run Finished task run for task with final state Mapped 2022 04 01 16 16 34 0400 INFO prefect TaskRunner | Task create flow run 0 Starting task run 2022 04 01 16 16 35 0400 ERROR prefect TaskRunner | Task create flow run 0 Exception encountered during task execution Traceback most recent call last File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect engine task runner py line 880 in get task run state logger=self logger File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect tasks prefect flow run py line 130 in create flow run flow = FlowView from flow name flow name project name=project name File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect backend flow py line 207 in from flow name order by= created EnumValue desc File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect backend flow py line 301 in query for flows f No results found while querying for flows where where r ValueError No results found while querying for flows where name eq l archived eq False project name eq L 2022 04 01 16 16 35 0400 INFO prefect TaskRunner | Task create flow run 0 Finished task run for task with final state Failed 2022 04 01 16 16 35 0400 INFO prefect TaskRunner | Task create flow run 1 Starting task run 2022 04 01 16 16 35 0400 ERROR prefect TaskRunner | Task create flow run 1 Exception encountered during task execution Traceback most recent call last File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect engine task runner py line 880 in get task run state logger=self logger File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect tasks prefect flow run py line 130 in create flow run flow = FlowView from flow name flow name project name=project name File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect backend flow py line 207 in from flow name order by= created EnumValue desc File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect backend flow py line 301 in query for flows f No results found while querying for flows where where r ValueError No results found while querying for flows where name eq i archived eq False project name eq i 2022 04 01 16 16 35 0400 INFO prefect TaskRunner | Task create flow run 1 Finished task run for task with final state Failed 2022 04 01 16 16 35 0400 INFO prefect TaskRunner | Task create flow run Starting task run 2022 04 01 16 16 35 0400 INFO prefect TaskRunner | Task create flow run Finished task run for task with final state TriggerFailed 2022 04 01 16 16 35 0400 INFO prefect TaskRunner | Task wait for flow run Starting task run 2022 04 01 16 16 35 0400 INFO prefect TaskRunner | Task wait for flow run Finished task run for task with final state TriggerFailed 2022 04 01 16 16 35 0400 INFO prefect FlowRunner | Flow run FAILED some reference tasks failed Process finished with exit code 0

Kevin Kho · Answer

This looks off because ```create flow run map flow name= livelots flow project name= LiveLots ETL ``` treats the strings as iterables so you end up mapping over the letters of the string Can you try with unmapped ```create flow run map flow name=unmapped livelots flow project name= LiveLots ETL ``` and using `from prefect import unmapped`

Patrick Tan · Answer

There is some success with using unmapped function The child flow ran twice with 2 set of parameters expected However they are run at same time with same flow id I expected running one at a time and also get some error message ```with Flow parent flow as flow filelist = get filelist task mapped flow run ids = create flow run map flow name=unmapped livelots flow project name=unmapped LiveLots ETL idempotency key=datetime now strftime m d Y H M S parameters= config file Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml bucket wp ahdata sandbox prefix dot scrapy live lots grays test live lots raw tsv config file Users patricktan bitbucket data live lots etl live lots etl batch config dev test yaml bucket wp ahdata sandbox prefix dot scrapy live lots Blackwell 20220325 180155 Blackwell DS=420 tsv wait for flow a = wait for flow run mapped flow run ids raise final state=True flow run ``` 2022 04 01 16 39 42 0400 INFO prefect FlowRunner | Beginning Flow run for parent flow 2022 04 01 16 39 42 0400 INFO prefect TaskRunner | Task create flow run Starting task run 2022 04 01 16 39 42 0400 INFO prefect TaskRunner | Task create flow run Finished task run for task with final state Mapped 2022 04 01 16 39 42 0400 INFO prefect TaskRunner | Task create flow run 0 Starting task run 2022 04 01 16 39 42 0400 INFO prefect create flow run 0 | Creating flow run 78d36f6c 53ba 4865 bf6a 8d88f2f6e16e livelots flow for flow livelots flow 2022 04 01 16 39 43 0400 INFO prefect create flow run 0 | Created flow run 78d36f6c 53ba 4865 bf6a 8d88f2f6e16e livelots flow 2022 04 01 16 39 43 0400 INFO prefect TaskRunner | Task create flow run 0 Finished task run for task with final state Success 2022 04 01 16 39 43 0400 INFO prefect TaskRunner | Task create flow run 1 Starting task run 2022 04 01 16 39 43 0400 INFO prefect create flow run 1 | Creating flow run 78d36f6c 53ba 4865 bf6a 8d88f2f6e16e livelots flow for flow livelots flow 2022 04 01 16 39 43 0400 INFO prefect create flow run 1 | Created flow run 78d36f6c 53ba 4865 bf6a 8d88f2f6e16e livelots flow 2022 04 01 16 39 43 0400 INFO prefect TaskRunner | Task create flow run 1 Finished task run for task with final state Success 2022 04 01 16 39 43 0400 INFO prefect TaskRunner | Task wait for flow run Starting task run 2022 04 01 16 39 44 0400 ERROR prefect TaskRunner | Task wait for flow run Exception encountered during task execution Traceback most recent call last File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect engine task runner py line 880 in get task run state logger=self logger File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect tasks prefect flow run py line 258 in wait for flow run flow run = FlowRunView from flow run id flow run id File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect backend flow run py line 566 in from flow run id flow run data = cls query for flow run where= id eq flow run id File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect backend flow run py line 608 in query for flow run result = client graphql flow run query File Users patricktan bitbucket data live lots etl venv livelotsetl lib python3 7 site packages prefect client client py line 473 in graphql raise ClientError result errors prefect exceptions ClientError path flow run message parsing UUID failed expected String but encountered Array extensions path $ selectionSet flow run args where id eq code parse failed exception message parsing UUID failed expected String but encountered Array 2022 04 01 16 39 44 0400 INFO prefect TaskRunner | Task wait for flow run Finished task run for task with final state Failed 2022 04 01 16 39 44 0400 INFO prefect FlowRunner | Flow run FAILED some reference tasks failed

Kevin Kho · Answer

Looking again I think the idempotency key is weird also because it s a string so you might need to supply something of type `List str `

Patrick Tan · Answer

so the number of idempotency key should be same as my parameters

Kevin Kho · Answer

yes for sure and they need to be distinct

Kevin Kho · Answer

`wait for flow run mapped flow run ids raise final state=True ` you need to map this too btw and use unmapped for the True

Kevin Kho · Answer

Because that task just takes in 1 id

Patrick Tan · Answer

Thanks for all your help not quite done but making progress Will reach out next week if there is still issue Have a good weekend