Looks like agent process loops in increasing intervals to look for available flow runs, which goes up to 10 secs max. This latency can be huge for workflows that don't receive frequent requests, however should to processed without delays as & when triggered. Is there a way to override this behaviour?

Dask based question this time - anyone know what the underlying issue might be if I’m getting a Dask KilledWorker error? Error occured while running a Flow with 20 mapped tasks where each one triggers an instance of another flow. File “/usr/local/lib/python3.6/site-packages/distributed/client.py”, line 1841, in _gather raise exception.with_traceback(traceback) distributed.scheduler.KilledWorker: (‘trigger_subflow-10-e6f77a46d8da4b4ca8643d4fee63d6cb’, <Worker ‘tcp://127.0.0.1:39261’, name: 1, memory: 0, processing: 2>)

And for my next question - is there a way to use the graphql API to stop/kill a flow run with a flow_id? I can see an update_state mutation, it’s just unclear whether that actually stops it..

I'm excited that targets have been added for file-based caching, but I am having trouble with a very simple example. I have a task that merely copies a file from one place to another. I want that task to only run if the file doesn't exist at the destination. The code below does not work.

Traceback (most recent call last):
  File "C:\Users\pblankenau\Anaconda3\envs\gis\lib\site-packages\prefect\engine\runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "C:\Users\pblankenau\Anaconda3\envs\gis\lib\site-packages\prefect\engine\task_runner.py", line 661, in check_target
    new_res = result.read(target.format(**prefect.context))
  File "C:\Users\pblankenau\Anaconda3\envs\gis\lib\site-packages\prefect\engine\results\local_result.py", line 79, in read
    new.value = cloudpickle.loads(f.read())
EOFError: Ran out of input

class Copy(Task):
    def run(self):
        src = Path(r"D:\test\dumb.txt")
        dst = Path(r"D:\test\copy_in_here\dumb.txt")
        shutil.copyfile(src, dst)
        return dst

with Flow("cp") as flow:
    dst = Path(r"D:\test\copy_in_here\dumb.txt")
    result = Copy(target=str(dst), checkpoint=True, result=LocalResult())()

flow.run()

Hello, after prefect 0.12 release I have this kind of bug. Can anyone help me?

Hey All, I have run into a roadblock. 1. I have spun up the containers for apollo, ui, scheduler, postgres, hasura, graphql, and myapp using the docker-compose from HERE 2. I am able to register the flows and am able to see the registered flows at localhost:8080 3. To execute the flows, I have started a agent docker container in the same network as the other containers with command

prefect agent start docker --network prefect-server

I am still not able to execute the flows from the UI. Is there anything I am doing wrong?

Hi all, thanks for making this amazing project for the world. A quick intro: I work at a computer vision company that works with Giga-pixel images and I am doing preliminary investigation for tools that can manage our machine learning workflows. Prefect looks very interesting. A question --- is there a web page where I can better understand the current level of adoption "in production systems" for Prefect? I read a couple blogposts from early users and (presumably) customers. That said, I wonder if there's one that'd give a more holistic, up-to-date view. Thank you.

I’m using

GCSResult

in my flow and getting errors from GCS (caused by the

GCSResults

object trying to store task results:

Unexpected error: SSLError(MaxRetryError("HTTPSConnectionPool(host='<http://storage.googleapis.com|storage.googleapis.com>', port=443): Max retries exceeded with url: /upload/storage/v1/b/prefect-bucket/o?uploadType=multipart (Caused by SSLError(OSError(24, 'Too many open files')))"))

The reason is (probably) that I have many mapped tasks, and they are running in parallel. It’s a 32-core machine, but I can’t really figure out how Prefect/Dask decides how many run in parallel. Sometimes it’s more than 32. The bigger problem here is that I think it caused the flow to get stuck, so I didn’t even have the indication for the failure and couldn’t go and restart it. Anyhow, any suggestions on how to overcome this? Or at the very least cause my flow to fail on such things so that I can restart? Does it have anything to do with the fact that these tasks have a retry option?

Is there a recommended way to create custom slack notifications? (For example - craft a message using data from a task) I have slack notifications working for task states, just wondering how best to customize

Where is the docs for how the DAG is generated? eg, if I have

a = task_a()
b = task_b(a)
task_c(upstream_tasks=[task_b])

This produces a strange looking DAG. I would expect that to produce a DAG like task_a ---> task_b ----> task_c

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?

Hi I apologize if this is not the appropriate channel, but I'm trying to get a paid plan running on prefect.io and it looks to be unexpectedly difficult for a cloud service.

Hi folks, I see the comparison between Prefect and Airflow, but is there any comparison between Prefect and DolphinScheduler? Thanks.

Is there a doc link that has list of all the environment variables?

Hi! What is the proper way to kick off a flow run while executing something? e.g. I have someone click a button on a website and I would like to async kick off the flow run on my Dask Environment. Just calling

flow.run()

is running synchronous here, right?

Is there a good way to have a task generate two lists (e.g. error_items, and a processed_items as output and map each list to a different downstream task?

Has anyone ran into this error when running a Flow (Prefect version 0.12.1) on a Kubernetes Agent:

[2020-07-01 13:40:32] DEBUG - prefect.CloudFlowRunner | Failed to retrieve flow state with error: AttributeError("'SSLSocket' object has no attribute 'connection'")

The Flow runs fine on my local machine, and the healthcheck passes, but when it's deployed to K8, it seems the CloudFlowRunner cannot write logs to Prefect Cloud and aborts the execution of the Flow.

Hey folks! Has there been given any thought on how to add conda support for the optional dependencies? Perhaps I can help?

I'm running into an error when trying to confirm my flow can be registered

is_serializable(flow, True

gives me some debugging output but the error isn't particularly revealing

subprocess.CalledProcessError: Command 'python /var/folders/vz/kjj4n5wj2bb7q26n0kz4v4jw0000gp/T/tmpr1ux82t1' returned non-zero exit status 1.

Is there a way to get more detail from the debugger?

in order to use the new

S3Result

instead of

S3ResultHandler

, do we need to have a PrefectSecret named

AWS_CREDENTIALS

? Previously with

S3ResultHandler

there was a kwarg for `aws_credentials_secret`(which allowed us to specify different credentials for different Flows), but that doesn't seem to be present on the new

S3Result

. Trying to figure out how to pass those credentials to

S3Result

Hi all, I'm evaluating Prefect and I'm wondering how it handles mutual exclusion or dynamic scheduling of flows based on the state of other flows. Basically, if flows A and B both need exclusive use of some resource, can I set up Prefect to know that B should pause until A completes?

Has anyone used prefect for running short lived docker containers?

Hi - I am trying to understand the error message from flow.register. Is the message indicating that the flow is not valid?

prefect.utilities.exceptions.ClientError: [{'path': ['create_flow_from_compressed_string'], 'message': '[{\'extensions\': {\'path\': \'$.variableValues\', \'code\': \'validation-failed\'}, \'message\': "no such type exists in the schema: \'flow_group_insert_input\'"}]', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

Thanks

Hi all. I want to use dependency between docker images. For example, I have 1 container

myapp

and another container which is actually the prefect

agent

that will be running the flows. I figured out the

agent

needs to have the code that is there in

myapp

. So, I set up the flow like below,

flow.storage = Docker(dockerfile='path_of_Dockerfile_inside_myapp')  #this will ensure that Dockerfile I used to build myapp will be used to build agent
flow.storage.build() #this step fails with the eroor: FileNotFound. I looked at the prefect code in the github repo and this should work, but fails
flow.register(labels=['some_label'])

Is there a gap in my understanding of how the Docker storage works?

Hello! We are using KubernetesJobEnvironment. From what I can tell, this is the process: click run -> agent creates job -> job creates job with our kubernetes job environment Is this correct? Secondly, it seems to take quite a lot of time to create our kubernetes job environment job. Is this how it always is or does it have to do with our configurations? If it is always this way, is there a way to speed it up?

Hi all, I'm currently performing some testing on the self-hosted version of prefect. I have a flow which contains a task that needs to be resumed manually in order to continue the execution of the flow. On the cloud version, it worked perfectly but on the self-hosted one, I could not find any way to resume this task. That means my flow remains in the running state. Does anyone know how to do that? Thanks in advance!

Hi we are trying to use

register()

to push a flow to a ECR (amazon) private registry, but we can't figure out how prefect picks up the credentials to do it. We know that with the command line

docker build

and

docker push

, you need to execute a special

docker login

with

aws ecr get-login...

command, but don't know how to pass this on to prefect for register(). We feel like we miss a step.

Hi all! I’m experimenting with porting an old Luigi dag to Prefect and wondering what the most idiomatic approach would be. (I asked a similar question on the Airflow Slack because I’m trying to wrap my head around the differences in approach.) My current process: • Parses the HTML of a webpage every hour and fetches a timestamp value indicating the last time a linked file was updated • If the ETL DAG has not been run for that timestamp, it downloads the file, does some processing, uploads to S3, etc. • If it has been run successfully for that timestamp, does nothing and checks again in another hour. Do I want to create a DAG that ends with a task that creates a Result target on success, and begin the DAG with a task that checks for the existence of that target? Or is there a more Prefecty approach to checking whether the DAG has been successfully run with a given parameter (in this case, a fetched external timestamp) and skip it if so?

Hello everyone. I'm running a POC with prefect in a very simple flow: read data from s3, standardize numbers and write result to s3. For the moment we're trying local backend server (and not prefect cloud version) and we're getting a strange behavior: everything runs fine when use

flow.run(executor=executor)

but not with

flow.register()

. We're using using the FargateCluster to create a new cluster for the flow and it's setup properyl with flow.register but no processing is done. I had asked this before a couple of days ago here but now I have a reproducible exemple. I'll provide the details in this thread. Thanks a lot

hey hey, I'm a bit stumped trying to implement a data ingestion pattern in prefect. Is it possible to access task parameter values when templating the result 'target' filename? I'm trying to have my results stored in a directory structure (aka 'hive' dir partitioning). Thanks for any insight