hi everyone 👋 If I wanted to try Prefect now for my company. Should I go with Prefect 1 or Prefect 2 ? Aiming to build something in production within the next 2 months.

I’m getting intermittent errors about a Prefect Secret not being set in the cloud for my tenant. It happens every once in a while, and when I re-run the flow, it succeeds. Any idea what is causing the problem and what I can do to make sure it stops being flaky?

Exception raised while calling state handlers: KeyError('The secret SLACK_PREFECT_NOTIFICATIONS_WEBHOOK_URL was not found.  Please ensure that it was set correctly in your tenant: <https://docs.prefect.io/orchestration/concepts/secrets.html>')

For idempotency keys: I want to use the Prefect templating, in conjunction with a runtime variable. Will

create_flow_run(..., idempotency_key="{date}-"+MY_VAR)

function as expected? It's hard to tell, because I don't know yet how Prefect compiles the string templating as part of tasks. Would I need to generate this string as a task result, and then pass the result to the kwarg?

I have questions on the

late runs

icon on cloud UI. However, cloud-UI shows neither success (green status) nor failure (red status) Is this a issue about docker-login ? or are there other possible causes?

Hi all - I am trying to get prefect to work on Kubernetes hosted on aws EKS. I can access the UI and it in turn is able to access the graphql pod. I am having an issue where whenever i create a project it fails. Can anyone help resolve this. Is it related to the RDS postgresql database I connected? How would I go about debugging this

Does anyone know how to register a flow with DockerRun and GitHub storage, but in the GitHub storage, instead of a python file point it to a ipynb. I was looking into Prefect ExecuteNotebook and didn't find anything useful to me. I also tried tagging the ipynb cells with the tag parameter, but I don't think that worked. It is not a huge deal but it would be nice to make the ipynb the GitHub storage instead of making a python file for every notebook

I am trying to test caching behavior. I've run a flow to completion, and then marked it as failed, so that I can restart it. I have a log that looks like this:

13:53:26 INFO CloudFlowRunner 
    Flow run SUCCESS: all reference tasks succeeded
13:55:13 INFO
    User marked flow run 
            as Failed
13:55:21 INFO
    null restarted this flow run
13:55:22
    INFO agent Submitted for execution: Container ID: 
        <id>
13:55:22 INFO
    S3 Downloading flow from s3://<key>
13:55:23 INFO
    S3 Flow successfully downloaded. ETag: <>, LastModified: <> VersionId: <>
13:55:23 INFO CloudFlowRunner
    Beginning Flow run for <pipeline>
13:55:23 INFO CloudFlowRunner
    Flow run SUCCESS: all reference tasks succeeded

I would expect the flow to re-run entirely. However, it doesn't look like it does (it just jumps straight into Success). Is this due to Prefect's default input caching? To be specific, this flow is a flow-of-flows. I have two subflows

Ok second question - trying to register a flow using python library. prefect_client = Client(api_server=prefect_url). Getting an error "You have not set an API key for authentication." How would I get an API Key?

Does anyone know the preferred way to get the state of a Flow after execution? I tried to write a terminal_state_handler but cannot find how to import the 'State' module. Is this even the correct approach or is there another way that is better?

Hi everyone! I just downloaded Prefect 2.0 and was wondering if it's possible to create a dependency between two different flows. The reason I'm asking is because I want each flow to run in a different conda environment so it needs to be on its own deployment, but I also want it to be possible for a flow to run depending on the state of another flow in a different environment. Any help would be really appreciated. Thanks!

How do I debug why my flow runs are sitting in submitted stage.

Hi, new here and apologies if this has been asked. I have tried searching this slack and online for answers and the closest thing I could find that was an answer was the Flow-of-flows post. The scenario that I'm trying to deal with is the following: 1. There are two flows (A and B). I maintain B and another project team maintains A. 2. I cannot run the flow A as this flow generates data for the company at large for multiple projects to use. 3. I need my flow to run after the latest successful flow run of A (this happens on a roughly weekly basis but can sometimes be delayed) with updated parameters. Given the above, what is the best pattern to implement in Prefect? I have written a function that queries the GraphQL API and can return the result from the final state of the latest successful flow run of flow A. Is using this result the best way to proceed? The other team is willing to change the way their flows are run and what data/metadata their flows generate. I just cannot set it up as a flow-of-flows because they do not want other teams running their flows.

hi, is there a way that we can turn off/on a flow schedule from running a CLI command?

Are the date/times in the Prefect format strings returned as UTC?

GraphQL API python client syntax question: I'm trying to query a flow run by filtering on the value of the parameters. I cannot figure out how to pass a json blob object to the query using python. When I use json.dumps method python produces a string and as a result I get a bad request error. Not sure how to create a jsonb type parameters object and I was hoping someone might be able to point me in the right direction.

This is a really silly question, but I can’t quite clue out the answer from Dask’s and Prefect’s documentation. My aim is to use Prefect to orchestrate flows with varying memory requirements using a Dask cluster. As an MVP I’ve set up an

ECSRun

flow using the

LocalDaskExecutor

with 30GB of RAM. For large jobs this flow OOMs and gets killed by the Prefect scheduler. My question is this: If I set up a Dask cluster to run these jobs would it gracefully handle memory issues? That is to say if I had 30GB of RAM in the cluster and a job that required 50GB would Dask OOM or would it simply run slower? Do I need to modify my code to use Dask dataframes or is there some smarts here I’m not quite across?

This might be a silly question, but should

labels

be specified in both running environments (docker runner or local runner) and remote server-side (i.e. cloud UI or server UI) ??

Hi! I have a question about using two schedules in the flow of flows construct: I posted it here https://stackoverflow.com/questions/71839812/prefect-loop-over-flows-two-schedules

Hello everyone I am a beginner with Prefect, in order to create a Data Pipeline, I want to connect it with a SQL DB that I have locally so I used the documentation: https://docs.prefect.io/api/latest /tasks/sql_server.html#sqlserverexecute My questions are: -In order to configure the connection between SQL and Prefect I use the classes and I put it all in a .py file and I do execution?? -If Yes, at the .py code level, do a pyodbc import? - If not, how can I do? -for the args data & **kwargs what should be put? -For the password of the user what should be added to indicate it pwd or other arg?

I'm having issue with multiple layers of mapped tasks running out of expected order 🧵

Hi! Is it possible to define a task by decorating a class method in orion with @ task? looking at orion docs I don't seem to find examples that use class methods. Is this discouraged by design somehow?

Hello! I'm exploring Prefect integration with AWS and trying to extract values from a AWS Parameter Store inside a task. From what I understand from docs, I can run the method in the AWSParametersManager class, however, I'm getting an error from the

defaults_from_attrs

helper decorator. What am I missing? 😅 (Code in the thread)

What is the best method to split task resources up in a Flow. For example if I want a task to connect to a large spark cluster to do preprocessing, while for my ML task I want it to get connected to a GPU instance.

I have a workflow that writes a file to S3 I then have a flow-of-flows, which runs this workflow via

create_flow_run

, with the

idempotency_key

set to this file path. Running the flow the first time, will write this key. Running it a second time, won't -- as expected, because of the idempotency behavior. What I don't understand is that the second flow run has every task state set to 'success' -- but wouldn't I expect them to be 'cached' or something else, indicating that the flow isn't being re-ran?

Hello, I am receiving the following pickle error while registering my flow

File "/usr/local/lib/python3.8/dist-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/usr/local/lib/python3.8/dist-packages/cloudpickle/cloudpickle_fast.py", line 602, in dump
    return Pickler.dump(self, obj)
TypeError: cannot pickle '_thread.lock' object

to make it simple it has just one task as below

@task(log_stdout=True)
def get_file_names():
    files = s3.Bucket(s3_bucket).objects.all()
    file_names = []
    for my_bucket_object in files:
        file_name = my_bucket_object.key
        regex = re.search(r".ctl", str(file_name))

        if regex is not None:
            file_names.append(file_name)

    return file_names

and flow

# flow to chain the tasks
with Flow("my_flow", storage=storage, schedule=schedule) as f:
    ctl_files = get_file_names()

any ideas why prefect is unable to serialize / pickle ???

Is it possible to “force cancel” a flow? The flow was running on an agent that died and did not emit any events later. I manually clicked the “cancel” button from the UI, and now the flow is stuck in a

Flow run is cancelling…

message for 2 weeks. Is there anything I can do to let prefect cloud know that the flow is actually dead and never succeeded?

Is there a recommended debug way of shelltasks non-zero exit code in the docker-runner ?

Best practices question: I’ve found that since using prefect, I’ve used

try/except/else/finally

less often than before (in my main code) because I figure prefect will catch (and log) things. I still find uses for it inside actual tasks/functions. Is this what other people have done or am I thinking about it the wrong way? Thanks!

My co-worker and I have different prefect cloud accounts. How can I grant permission to my co-worker to view my flow run, or execute flow I registered?

Hi, I tried to create an VM from Prefect Agent in Azure Marketplace. The API key generated in Prefect cloud is 40 characters. But the Prefect Agent template requires the key length should be 1-22 characters. Can Prefect team help me to solve this issue?