Question: if I have a task I want to perform over all the lines in a file, but have each task run on a line depend on the next (I don't want to continue iterating if a previous line fails), whats the best prefect way to do this? for right now I'm doing:

with Flow('file') as flow: 
    name = Parameter('name')    
    lines = read_file(name=name)
    for l in lines:
        do_line(l)
    finish()

I know theres a prefect

Map

functionality, but I didn't see an easy way to make it so each task for each line depends on its previous line task in an iterative manner

Hey Michael! This is an interesting question - if you use the

LocalExecutor

with mapping, then your tasks will execute sequentially as you expect. I’m going to keep thinking about this though - I am intrigued by the idea of dynamic dependencies on dynamic tasks

yeah I want to use the dask executor so that doesn't really help unfortunately 😞 what I'm doing is creating a for loop and initiliazing tasks based on the number of lines in a file, kinda like above, is that not possible?

All Prefect tasks represent deferred computation, so it is possible but only if you read the file at Flow build time, not at runtime

Could you could use

Task.map

, and then

Flow.add_edge

to introduce new execution dependencies between the mapped tasks?

You can’t add edges to a Flow at runtime, because those edges are used when traversing the computational graph

dang 😞 so I'll need a flow constructor based on the size of the file? that messes up the whole flow unfortuantely 😞 the other option is to just make a REALLY big task which is less robust but ah well

out of curiosity - why do you want to use the dask executor for this? The Flow you appear to be describing doesn’t seem to benefit from parallelism

I'm running the same flow for different files, where no files depend on each other, so lines across different files are only dependent within that same file. These "file executions" can all happen in parallel, so i want to run them across dask

each flow run will occur independently / in parallel with other flow runs, so you don’t need dask to parallelize at the flow run level

will the localexecutor automatically parrallelize flows?

Executors don’t know about Flows. Agents submit Flows to run in whatever execution environment (subprocesses / k8s / fargate / etc). These submissions occur in parallel. Executors are configured on each individual flow - executors are responsible for running the tasks within a single flow.

hmmm ok seems like localexecutors may be the better way to go, but the only other issue is that I want the std in/out/error of my underlying python functions I call in these tasks, and dask makes this REALLY easy, I haven't found a really good way to do that with a local executor

No worries at all! I think I’m a bit confused - the local executor doesn’t manipulate output in any way so it should be much easier to see the output from your tasks in the local executor vs. the dask executor

I'm not running an agent

Ah then all your stdout / stderr from your tasks should already be displayed

yeah but its not recorded very easily without doing:

python3 flows.py > out.log

Yea it sounds like you should be using a Prefect backend (either server or Cloud) to manage this and persist your logs

yeah lol I should be XD, but another issue is I'm trying to avoid using docker with my project, as I said its a little complicated

if you use Cloud you don’t need to use Docker

i really like prefect server, its really nice, but its not very lightweight 😕 (versus just only depending on python code)

gotcha - for what it’s worth we have a lot of highly secure clients on Cloud if you’d like to have a discussion on security we’re open to it, but I know these things can be complicated so no pressure!

yeah thanks! atm i don't think it'd be possible, but I think its something I'd be interested in the future