Hello everyone. I'm trying to run a simple task using prefect server and I'm having some trouble with it. The idea is to read a parquet dataset, standardize the feature columns and write the result to a csv file. I'm using dask and dask-ml to process the data and code looks like:

with Flow("standardization") as flow:
        df = read_feats(bucket, file_pattern) # a task that returns a dask dataframe using dd.read_parquet(...)
        scaled_df = scale(df) # prefect task that does the standardization using dask-ml StandardScaler
        write_csv_to_s3(scaled_df, bucket, output_file) # this will write the dask dataframe as a csv to an s3 bucket

This is a very simple POC just to get things running and feel prefect in action. I'm using version 0.12. Everything runs smoothly if I run the flow locally using flow.run with a remote DaskExecutor in a FargateCluster (using dask-cloudprovider):

executor = DaskExecutor(...) # cluster parameters setting fargatecluster from cloudprovider and cluster_kwargs
flow.run(executor=executor)

I start having problem if I try to run it using a local prefect agent connected to a local prefect server after registering it:

executor = DaskExecutor(...) # same parameters as before
flow.environment = LocalEnvironment(executor=executor)
flow.register()

The agent will deploy the flow, it will create FargateCluster normally and we can see the tasks registered at Dask UI Tasks stream but no processing actually happens. It desearializes the first task and do nothing, then does the same with the second and third tasks. Any idea what am I doing wrong here?

Sorry, having a bit of trouble parsing your question. You're saying things run fine when you do:

flow.run(executor=executor)

but don't when you do:

flow.environment = LocalEnvironment(executor=executor)
flow.register()

with the same executor configuration?

Exactly

Hmmm, the tasks are arriving, so you have workers and things are connected, but you're saying the tasks do nothing? From looking at the above task plot in the dask UI, you're having tasks that do complete. How do you know they're doing nothing?

They run too fast and the final file is not written

Is your flow successful, or is the task failing?

Flow just keeps running forever

Hmmm, that is odd. Can you get the logs from your dask workers?

From what I see all tasks are "pending"

I suspect this is an environment error

I just tried doing that but the scheduler and worker were killed before I could do it

If the cluster shuts down, that means the

FlowRunner

did complete.

Let me try that.

Sorry, try using a

DaskExecutor()

(runs a local distributed cluster) - the

LocalDaskExecutor

uses the local threaded scheduler which is different. I suspect things will work the same, but this is a better proxy for what you're doing.

It works just fine with the DaskExecutor() as well

Ok. Hmmm. So a few thoughts on what might be happening: • The images your cloud provider cluster is running with don't have all your dependencies available, so something weird is happening. • The configuration to talk to prefect cloud isn't getting properly set in the cloudprovider cluster, so calls to the prefect api are failing, resulting in weird behavior. I'd expect more informative and less weird errors in both these cases, but without logs it's hard to check.

I believe it's not the image problem as we generated a docker image with all dependencies. And we also get to run on the fargatecluster when calling flow.run(executor=executor)

They say nothing as in nothing unusual? Or they say nothing at all? There should be logs, especially since you're seeing tasks run.

I see nothing unusual and nothing about tasks in general. Maybe because they are at info level?

Hmmm. Ok. Are you running with prefect cloud or with prefect server?

I'm trying it with prefect server now

Ok. You might try to get your prefect logs for this flow run using the CLI.

prefect get logs --id FLOW_RUN_ID

Just logs from dask

And that's it? Hmmmm

That's all.

I'm a bit stuck on this. If you can provide a reproducible example (perhaps using some dummy tasks) I'd be happy to take a look in more detail. As is this isn't much to go on.

No problem. I'll try to create a reproducible exemple with dummy data and will get back to you.