Hi, I cannot seem to get my flow to run on a LocalDaskEnvironment from the cloud UI. When I use

flow.run()

locally it seems to work. Here is the code snippet I am using for registering my flow. I am using

Docker

storage.

if args.base_image:
        storage = Docker(
                base_image=args.base_image,
                registry_url=args.registry_url,
                local_image=args.local_image)
        for flow in flows:
            path = storage.add_flow(flow)
            print(f"storing flow {flow.name} at {path} in the image.")
        storage = storage.build()
        for flow in flows:
            flow.storage = storage
            flow.environment = LocalEnvironment(
                    executor=LocalDaskExecutor(),
                    labels=args.labels)
            flow.register(project_name=args.project, build=False)

Am I doing something wrong? I am on the developer version of the cloud backend. EDIT: To be more precise, when I run

flow.run()

I see concurrency on the mapped tasks, but not when using the cloud API (even for a single flow).

Hi @karteekaddanki, try using the

RemoteDaskEnvironment

instead of the local environment, like this:

from prefect.environments import RemoteDaskEnvironment
###
###
flow.environment = RemoteDaskEnvironment(address="<tcp://YOUR_CLUSTER_ADDRESS>", labels=args.labels)

Thanks @nicholas. I am a bit confused. I do not have a remote cluster. Even if I did start the cluster on the same machine, I'd have to say

<tcp://127.0.0.1>

in my configuration. Any particular reason why the LocalEnvironment isn't concurrent with LocalDaskExecutor?