Hello, maybe this is kind of a silly question but - assuming we have a big CSV that we then split into multiple CSVs (as a task), and then, we wanna run a set of tasks for each of these “mini-CSVs” (either in parallel or sequentially, probably sequentially though) — we’re trying to understand what’s the beach approach here (continued in thread…)

you would need to use iterated mapping https://docs.prefect.io/core/concepts/mapping.html#iterated-mapping but I agree this is challenging if some processes are not Python. Rather than env var, perhaps you can push that info to somve KV Store, grab that info in a Python process/Prefect task and pass to downstream tasks? just some ideas

hmm, not sure - what is the KV store meant to solve?i dont have a problem altering the env for the shell task (this could be done at runtime by passing it an

env

param from what i understand) - i just dont understand how to combine multiple prefect tasks into “one” that can then be mapped ?

You don't have to combine them into a single task, you can juste "chain" mapped tasks together, creating some sort of "pipeline" for each of your initial objects.

@Sylvain Hazard oh, interesting, is there an example for this?

Yes, this is a good example

so i guess now i just need to figure out a way of wrapping a

ShellTask

in a way that i can convert a name of a file (e.g.

my_chunk3.csv

) to an env param

You could maybe add a task that generates the env param name from the file name ?

for example:

saved_main_csv = save_as_csv(accounts_data)
chunk_names = split_csv(file_name="/tmp/myin.csv", rowsize=5000, upstream_tasks=saved_main_csv)
run_output = sh_task(command='npm run partial_enrich').map(chunk_names)

^^ this wouldn’t work because i don’t think

sh_task

can be mapped this way, and even if it could, just giving it the list of filenames wouldn’t inject them as an env param

try it and you'll know 😉 to make it sequential, you would need to attach LocalExecutor to your flow

@Anna Geller hmm - im not sure if it works or not cause the mapped tasks are stuck in the

Mapped

state… is that related somehow to not having a

LocalExecutor

?

What was your executor again? I scrolled through but didn’t see?

i didn’t define any specifically, later, based on Anna’s suggestion i explicitly set it to

LocalExecutor

it doesn’t seem to make any difference, the tasks won’t progress from a

mapped

state and don’t seem to generate any “child tasks”

I can’t really see anything wrong, and we normally see this on LocalDaskExecutor. Are you open to trying task looping if you need this to be iterative anyway? Also, I think you can try something like:

for chunk in chunk_names:
    wrapped_enrich_shell_task.run(chunk)

so that there is no Prefect Flow and you can see if the loop still hangs

but there is no loop, it just doesn’t create any child tasks i’d very much like to understand why it doesn’t work since it seems pretty straightforward (and surely we’ll need mapping elsewhere anyway)

The map is basically like a for loop for LocalExecutor. I’m suggesting to use it so we can see if it hangs when running outside of a Flow context

how can i run it outside of a flow context though? not sure i follow

If you have a task,

@task
def mytask():
    ...

mytask.run()

will just call the Python function under the hood. No

Flow

object needed. This way we can debug the Python under the hood

are u saying that instead of mapping i move the for loop inside the wrapped task?

Cuz I think something might be getting stuck there

this is the wrapped shell task:

@task(log_stdout=True)
def wrapped_enrich_shell_task(input_file_name):
    logger = prefect.context.get('logger')
    <http://logger.info|logger.info>(f"starting to work on file {input_file_name}")
    output = sh_task.run(command='npm run partial_enrich', env={
        "LE_INPUT_FILE": input_file_name,
        "LE_OUTPUT_FILE": 'enriched-' + input_file_name
    })
    return 'enriched-' + input_file_name

So I am proposing instead of:

mytask.map(..)

we try the for loop

ok so i move the loop inside this task you mean?

No no. Let me make an example

like so?

@task(log_stdout=True)
def wrapped_enrich_shell_task(input_file_names):
    logger = prefect.context.get('logger')
    output_file_names = []
    for input_file_name in input_file_names:
        <http://logger.info|logger.info>(f"starting to work on file {input_file_name}")
        output = sh_task.run(command='npm run partial_enrich', env={
            "LE_INPUT_FILE": input_file_name,
            "LE_OUTPUT_FILE": 'enriched-' + input_file_name
        })
        output_file_names.append('enriched-' + input_file_name)
    return output_file_names

If you have this:

from prefect import Flow, task

@task
def mytask(x):
    return 1

and have a Flow:

with Flow("...") as flow:
    mytask.map([1,2,3,4])

take out the Flow and do:

for x in [1,2,3,4]:
    mytask(x)

so in your case,

for chunk in chunk_names:
    wrapped_enrich_shell_task(chunk)

and just comment out the flow for now

i can’t run this task out of context, there’s too many stuff it needs to run, env vars, the input from snowflake, etc.

I think the issue might be the multiple shell tasks though. It’s ok if you can’t run the task out of Flow context, can we try then just commenting out the content of the

wrapped_enrich_shell_task

and just running something like

echo something

and then mapping over that to see if that can reproduce it?

sure

Also what happens on LocalDaskExecutor?

i never tried it

Yes that would help at least isolate if the behavior is the shell task or potentially related to the

npm run

same thing:

Ok that’s helpful. I can try to replicate on my end

but i understood that

LocalDaskExecutor

will execute in parallel, im not interested in parallel excution. i can try it just for the sport but it’s not what im interested in

Yeah that would help

same thing:

Trying to replicate with this:

from prefect import Flow, task
from prefect.tasks.shell import ShellTask

sh_task = ShellTask(helper_script="cd /usr/src/app",
                    log_stderr=True,
                    return_all=True,
                    stream_output=True,
                    log_stdout=False)

@task
def wrapped_enrich_shell_task2(input_file_name):
    sh_task.run(command=f"echo {input_file_name}")
    return 'enriched-' + input_file_name

@task
def get_file_names():
    return ['a.txt','b.txt','c.txt','d.txt']


with Flow("..") as flow:
    names = get_file_names()
    wrapped_enrich_shell_task2.map(names)

flow.run()

same thing

I get something similar, though it seems the mapped task runs do run

ok

No. I think I know one issue

looks better now though:

Can you try:

from prefect import unmapped
run_output_filenames = wrapped_enrich_shell_task2.map(chunk_names, upstream_tasks=[unmapped(touch_output)])

(i did some other changes thoguh, like remove the

upstream_tasks

and the other upstream task)

I will explain if it ends up helping

yep, looks good:

Ok I can draft up an example to show you the difference

hmm ok sure, thanks

LocalExecutor is serial execution, but order is not guaranteed. Only looping guarantees order. Checkpoint works, but there is a caveat that you need distinct results for each of the mapped tasks because it checks if the result exists. So you need to use templating to give them distinct names.

i thought setting

upstream

only tells it the dependency though, not that it influences what it tries to run on

So if the upstream is a list also. The assumption is to pair the downstream execution to each upstream in an element-wise fashion so that we can do depth first execution as opposed to breadth first.

huh, interesting

These are very tricky, but I believe it has to be that way to support the depth first execution of consecutive mapped calls.

regarding templating - i think it would be easy in our case because we split the big CSV into smaller ones where each one is called for example 5000A, 5000B, and so on… so that could be in the task name too, right?

Yes that can be in the Task Run name. It will work with any Result we have, but the PrefectResult has a size limit I think because that is stored in our database compared to something like S3 result which is a file in a bucket you own.

but we have to have it set to something other than local i presume?

LocalResult can work too with the checkpointing. But if you restart the Flow, you might lose those files if you use a container and it restarts unless you use some kind of of hard disk?

we upload the results to S3, i guess what you’re saying is that we’ll lose the data if the flow somehow dies in the middle before all results were persisted?

If you are relying on container storage which is ephemeral right?

while running, yes, but the assumption is that pushing it to s3 upon success would persist it

I think this will be fine, but the assumption is that you never have to restart between saving locally and uploading to s3. If a file is saved but the upload fails, then you won’t have it upon the Flow restart because the container died right?

rather than - run all , then read all, then push all

I get that, but the upload can fail right? That would require a restart

presumably, yes. but im not worried about that. it never failed for us 🙂 and i can add retries

But I think anyway successful tasks are not re-run? We can’t enforce depth first execution though so the risk here is that if you have

Save File A -> Save File B -> Save File C -> Upload File A -> Upload File B -> Upload File C

and then

Save File C

fails, you lose the work of A and B and upon restart, it’s not there to upload anymore. This feels a lot simpler if you directly use the S3Result

wait so depth first is not enforced?

For Local and LocalDask, not enforced. For DaskExecutor, highly preferred but not enforced

interesting. ok thanks, i will play around with it a bit, much appreciated 🙏

Of course!

k, just tested it out with an S3Result and added random failures to the (mock) shell task, tried a restart and it did pick up well from the failures and finished everything (in the end i had to set an S3Result for the entire flow, cause otherwise the mapped task has nothing to work on, which i kind of missed earlier — although i suppose i can also turn off checkpointing for the earlier steps) i think the only thing i dislike is that it just throws everything to the bucket unless i manually tell it to go to a specific dir (now the top level of my entire prefect bucket is filled with persisted task results…) but at least it’s working 😄

wohoo! those 100 messages paid off 🎉 😄

hug