Hey folks! Any idea why I'm able to get parallelism in my flow when I pass the

LocalDaskExecutor()

to the

flow.run(..

method but when I pass it to

LocalEnvironment(executor=LocalDaskExecutor())

and then pass that to the flow constructor like this:

with Flow("test", schedule=schedule, environment=LocalEnvironment(executor=LocalDaskExecutor())) as flow:

I only get one task running at once. Here's my test code:

@task
def generate():
    return [x for x in range(0,40)]

@task
def log_sleep(x):
    logger = prefect.context.get('logger')
    time.sleep(5 + x)
    <http://logger.info|logger.info>(x)
    return x * x

@task
def collect(lst):
    logger = prefect.context.get('logger')
    <http://logger.info|logger.info>(lst)

schedule = Schedule(clocks=[DatesClock([pendulum.now() + timedelta(seconds=5)])])

with Flow("test", schedule=schedule, environment=LocalEnvironment(executor=LocalDaskExecutor())) as flow:

    nums = generate()

    results = log_sleep.map(nums)

    x = collect(results)

flow.run()

Hey Chris! Yea this is a common confusion that we hope to fix up soon — the

environment

that you configure for your Flow is only used when running against a Prefect backend, so when you call

flow.run

the default

LocalExecutor

is used

ah ok - what would be the easiest way to test this locally? I'm trying to figure out why my flows are running slow on my server (everything is in a docker container on GCP). the flows get registered and then run from prefect cloud

What sort of slowdown are you seeing?

mapped tasks running one by one instead of in parallel

Interesting - are they extremely fast running tasks?

not particularly. it's a multi-level mapped flow

OK yup this all looks good to me

none as far as I can tell

No your code that I can see looks good and standard to me, I suspect this isn’t a Prefect-specific issue but rather a Dask issue. Can you SSH into your server and perform the same local run that you did above to see what happens?

yeah let me try

So if you set the following two environment variables that should make it easier:

PREFECT__CLOUD__AUTH_TOKEN=YYYY
PREFECT__CLOUD__USE_LOCAL_SECRETS=false

ok I think I might know what's going on - I think it might be using an older image version on the server. thanks - will keep digging in and let you know

ok great, sounds good!

yup - that's what it was. machine had booted and pulled an older version of the image which wasn't configured. nothing like spending an hour on a

turn it off and on again

problem 😛

Haha classic, glad you figured it out

thanks again for your help - understanding the environment/runtime executor was definitely helpful

Anytime!