Hi, how can you delete a storage configuration with the cli utility in prefect 2.0?

If a want to run a large backfill on k8s using the framework defined here https://discourse.prefect.io/t/i-want-to-use-parameter-values-for-backfilling-but-prefe[…]ter-defaults-provided-at-registration-how-to-solve-that/49 - is there a way to batch things up so I am only ever running say 10 jobs at a time?

Hey, 1. What is the best practice for running Tasks (not flows) over Kubernetes ? I think about using DaskExecuter and deploying Dask over Kubernetes cluster. Just want to make sure if that is the correct way 🙂 2. What is the best practice for managing CI/CD of flows? I wish to deploy and run branch specific flows and delete them upon merge Thanks!

Hey, I have a Secret AWS_CREDENTIALS in Prefect Cloud (1.0) following this format:

{
  "ACCESS_KEY": "abcdef",
  "SECRET_ACCESS_KEY": "ghijklmn"
}

If I'm using pandas to read a file in S3:

df = pd.read_csv(f"s3://{s3_bucket_name}/{filename}")

Should I need to pass the credentials as a param to the function read_csv? Or are they read automatically from the Cloud? Currently I'm getting this error: "botocore.exceptions.ClientError: An error occurred (403) when calling the HeadObject operation: Forbidden" Thanks!

Hi, When I run a task locally using

flow.run()

, it works just fine. However, I’m running a local backend server and when I try to register the task using

flow.register()

, I get this error:

File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 736, in save_tuple
    save(element)
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 496, in save
    rv = reduce(self.proto)
  File "/Users/shriramholla/.pyenvs/python3.6/lib/python3.6/site-packages/absl/flags/_flagvalues.py", line 677, in __getstate__
    raise TypeError("can't pickle FlagValues")
TypeError: can't pickle FlagValues

Note: Nothing is being passed into or being returned by any task

Hi! I would like to ask about a case with multiple return values from a Task. When I specify the flow edge manually by using

flow.add_edge(_upstream_task_, _downstream_task_, _*key*_)

is there any way to specify which of the multiple return values from the upstream task should pass to the downstream task?

Hello! I’m trying to figure out why my local runs differ from prefect cloud. I expect that my task will handle exception, and it perfectly works local, but in cloud my flow fails with strange error message “‘traceback’ object is not subscriptable”. How can I fix this?

Hey guys, what is the best way to copy a flow? Just use .copy() or deepcopy?

This message was deleted.

Hi #prefect-community, does anyone have any experience running prefect agent in Windows Server?

How do I link two ShellTasks so that they run sequentially?

Prefect 2.0 extra loggers does not seem to work when used with custom libraries

Hi everyone! I'm facing issues with prefect logger, as I'm using ThreadPoolExecutor, I want to ensure that future is completed successfully or at least know the reason of failure. The flow and task with parallel execution looks like this. The deal is that when running locally, I'm able to see the logs, but couldn't see them on Prefect UI, though I'm able to explore other logs outside this task. I'm using

prefect.context.get("logger")

as logger. Could anyone guess how to fix this? Thanks

Hi everyone. This morning I encountered problem when running prefect workflow. the log say : RuntimeError: Missing dependency kubectl. Please install kubectl following the instructions for your OS. This never happened before, I try to update kubectl at the worker cluster and not helping. I'm using dask on kubernetes for executor. any idea?

Hi i have multiple agents running on Prefect. Incase any agent goes down (turns yellow/red) can we get slack alert to get notified?

Hi Everyone. I have deployed a prefect server with a Kubernetes agent using the Helm chart in a EKS cluster. Now I'm kind of confused how to register the flow? Where is the Python code supposed to be in Kubernetes?

Hi, how can I stop a running flow-run either with the web ui or the cli?

hi community! is there a way to look-up a flow's id using the flow name/project name via the client? this suggests you can via the cli

Hi , What’s best practice to get the value of a prefect Parameter inside a flow?

Hi :-) StartFlowRun() with map() not creating unique runs. I have a wrapper flow, that creates sub-flow runs:

flow_run = StartFlowRun(project_name=PREFECT_PROJECT, flow_name=SUB_FLOW)

, that I execute with

flow_run.map(parameters=parameters, run_name=run_name)

. (Here

parameters

and

run_name

are generated by other

map()

tasks, and of same length.) When I run the wrapper task locally, targetting the sub-task in Prefect cloud, I get:

└── 12:48:22 | DEBUG   | Flow Run 7951f47c-1ddf-4e33-b913-a2698a89e4c9 created.
└── 12:48:22 | DEBUG   | Creating link artifact for Flow Run 7951f47c-1ddf-4e33-b913-a2698a89e4c9.
└── 12:48:23 | INFO    | Flow Run: <https://cloud.prefect.io/.../flow-run/7951f47c-1ddf-4e33-b913-a2698a89e4c9>
...
└── 12:48:23 | DEBUG   | Flow Run 7951f47c-1ddf-4e33-b913-a2698a89e4c9 created.
└── 12:48:23 | DEBUG   | Creating link artifact for Flow Run 7951f47c-1ddf-4e33-b913-a2698a89e4c9.
└── 12:48:24 | INFO    | Flow Run: <https://cloud.prefect.io/.../flow-run/7951f47c-1ddf-4e33-b913-a2698a89e4c9>
...

Note the same GUIDs!!! When we visit the cloud UI, only the first mapped task is created! If the wrapper task is also running in cloud, it works as expected. I'm not sure how to create a minimal reproducible example for this, since

StartFlowRun

can't trigger a local flow (I don't use prefect server, only prefect cloud).

Hi, we migrated from prefect 1.0 to prefect 2.0. With this migration we are experiencing huge performance issues running the same flows as before. Interestingly, when keyboard interrupting a prefect flow on the CLI the task continues to run - with better performance. It seems this keyboard interrupt exits a prefect wrapper that is responsible for the performance issues. Attached is a screenshot of the error code when interrupting a prefect flow. Anybody experiences similar performance issues after the migration?

Hi All, I am trying out Prefect 2 for the first time I have a flow that imports some functions from another (local) python module In prefect 1, we were able to do something like

storage.files.update(THE_FILE)

to bring in the dependency What’s the recommended approach to achieve something similar in prefect 2

Hi, there is this LOOP construct in 1.2. I was wondering when that will become available in 2.0. Thanks!

Hey there ! I have a mapped task for uploading files through an API call and I'd like to avoid having to authenticate to the API for every task but my client is not JSON serializable and cannot be passed as a parameter. I'm pretty sure I've seen a way to properly deal with this before in the docs but can't seem to find it again. Any idea or advice ?

Hi! Is there a way to start an agent via a python script instead of the command line?

Hi all, would anyone know why a flow would be running tasks in a different order than expected? It seems that the flow that I am running locally is constantly changing the order of how tasks are run. I haven't had this issue until recently

Hey! I'm working with Prefect in Azure using K8s, ACR and Prefect Cloud. Is there a setup for affinity or taints/tolerations? I'm trying to put tolerations in the KubernetesRun template and the DaskExecutor as well. It works for tolerations, but on the DaskExecutor, it gives this error:

RuntimeError: Missing dependency kubectl. Please install kubectl following the instructions for your OS.

Does anyone know about this error with K8s tolerations?

Hi! I know I can delete a flow using

from prefect.client import Client

flow_id = "..."

client = Client()
client.graphql(
    """
    mutation {
    delete_flow(input: {flow_id: "%s"}) {
        success
    }
    }
    """ % flow_id
)

But how can I delete all versions of a given flow? It looks like the Flow Group ID is constant over the flow versions. I tried to replaced it into above snippet but didn't work.

Hi, I run some flows directly from vs code, and I stopped them manually from IDE before they finished. It was a few hours ago. In meantime I restarted computer and UI still shows me them as running. Is that normal? How can I clean them?

Hi, running this in container (AWS ECS) failed but works running from desktop computer