Hi all, Working on a simple Flow but my Tasks keep stuck on "Pending”. Logs say

Failed to load and execute Flow's environment: ValueError('Flow is not contained in this Storage')

My setup is on a AWS EC2 machine and I am running the Flow with a Docker Agent, as per below.

from prefect import Parameter, Flow, task
from prefect.storage import GitHub
from prefect.run_configs.docker import DockerRun


@task(log_stdout=True)
def test(e):
    return e.get("body")


storage = GitHub(repo=repo,
            path="flows/flow.py",
            access_token_secret="GITHUB_ACCESS_TOKEN")

run_config = DockerRun(labels=["L1", "L2", "L3"],
                       env={"EXTRA_PIP_PACKAGES": "zcrmsdk zohocrm-prefect-tasks"})

with Flow(name="Name", storage=storage, run_config=run_config) as flow:
    # Pipeline parameters
    event = Parameter('event', required=True)
    test(event)

Any idea about what I am doing wrong?

Short version: Is there a way to set retries to use exponential backoff? Long version: The documentation at https://docs.prefect.io/core/tutorial/04-handling-failure.html has retry options at set intervals, but network issues (often the cause of my failures) generally work better with exponential backoff (https://dzone.com/articles/understanding-retry-pattern-with-exponential-back). Is there a way to do this currently?

Hi guys, My problem is following: I’d like to pass the logs (standard python logger) from the submodules to prefect logger ( so it can be displayed on the server). I understood, that it can be done with

run config

in flow (I do the local run with the local agent):

run_config=UniversalRun(env={
    "PREFECT__LOGGING__EXTRA_LOGGERS": '["lib1", "lib2"]',
    "PREFECT__LOGGING__LEVEL": "DEBUG",
})

I can ONLY observe

info

level from my libs even if I try to switch on

debug

level! (However the prefect logger shows the information related to

CloudFlowRunner

from

debug

level without any problem). What do I wrong?

Can you manually specify an upstream task tied to a renamed task? Example, if

@task(name = "Renamed Transform", log_stdout=True, checkpoint=False)

, can I set upstream_task = ["Renamed Transform"]) in a downstream task?

Hi

I have a regarding resuming failed flows. I hope you can help me.

Suppose I have a flow that consists of 4 tasks that run run one after the other, e.g.,

A -> B -> C -> D

. Now suppose I run the flow and tasks

A

and

B

complete, but

C

fails because of a bug. I therefore want to fix the code in task C and redeploy the flow.

Tasks

A

&

B

are expensive to run, so I would rather not re-run them from scratch. Is it possible to re-run the flow with the newly deployed code from task

C

onwards?

For some reason I was expecting the answer to be obvious, but now that I try it I can't seem to find a solution

Hey y'all I just wanted to say that today I was getting '*Unauthorized. Invalid Prefect Cloud API key.*' when I ran

prefect auth login --key <MY_KEY>

even though my key was valid and authorized. I created so many keys later on still didn't work ....and in the end it turned out that I had already logged in with another key couple days ago. So i had to run

prefect auth logout

before logging in with the new key .

I feel like this is a Noob question but I can't find an answer. Is there a way to make prefect server restart automatically after reboots? The documentation mentions using supervisord for agents on this, but there don't seem to be any equivalent instructions for server. The closest I found was another thread on here that seems to use

prefect server config

to generate a docker compose file, and I'm partway down the path of trying to set that up to auto-start (still working through https://github.com/flavienbwk/prefect-docker-compose, which was linked in it), but so far it's not working and it seems like there should be a simpler answer here. Docker has restart policies already (https://docs.docker.com/config/containers/start-containers-automatically/), is there not some flag I can pass to

prefect server start

to use them?

What's the best way to see the first 10 rows of data I've transformed in prefect without running in the cloud? I can manually embed my credentials temporarily and want to do something like this and see the result in the terminal:

flow = Flow("myflow")
with flow:
    df = extract(stuff)
print(df.head(10)) # This does not work

Evaluation of Orion (Alpha5 as well as the Dev-Version from GH lead to the same findings) We have certain questions concerning Prefect Orion (Prefect Alpha5) • There seems to be no .result() to flows any more. So how does one get the result of a subflow back? • (Sub)-Flow input parameters have to be JSON serializable. Therefore any object/instance parameters are no longer possible. Not even a simple numpy array would work. Why not cloudpickle? • Failed Tasks are not displayed in Radar. • There is no start/stop/restart functionality for Flows. Will there be checkpointing in Orion? • Each parallelized task instance produces a new box in the radar. In Prefect this was much better visualized. • The radar is an interesting new view on the graph. It can be of use. But what we really would like to have is the possibility to allign a certain flow/task at a given coordinate. This is most interesting if one have production code that runs the same flows 24/7 for the next years. The graphs is then frozen. • There seems to be no way in Orion to store/retrieve self serialized results any more? The Docu states that results were stored in the SQLiteDB. Is this really the future approach? Usual payloads are in the GB range. • Even for trivial examples (inc/dec example with a flow of flows) with local parallelism (DaskExecutor) we get concurrent access to the SQLiteDB resulting in failures. These faillures seem to heal themselves (retries were made that conclude positive). These collisions are not easily reproduced which hints at a runtime (async probably) problem. • In Prefect there were two APIs to address your problem. Functional and Imperative (Baseclasses). The imperative API is not covered with a single word in the Orion documentation. We would really like to have an imperative API in Orion. What are your plans for the APIs in Orion? Sorry for so much criticism. Orion is quite impressive. The Engine does a good job in auto-parallel. With a few decorators one can optimize its code with no cost. But Orion is IMHO quite far from production. Volker

Question about github storage

Hello everyone, in prefect, what's the idiom to flatten back the results of multiple parallel tasks from apply_map()? I've tried various combinations of syntax and unfortunately I can't seem to get it to work.

QQ how can I assert the state of a dependent flow is successful? I'm running the flow using StartFlowRun

Does prefect have a drag-and-drop-based UI,? I want to use it for network operators. They can’t code.

Hi everyone — I’m struggling understanding how prefect behaves when I redeploy stuff. The way it works now is that 1 git repo = 1 prefect Project containing multiple flows. We have a dockerfile whose endpoint simply registers all the flows and then starts a LocalAgent. This is deployed as an ECS Service. When we deploy, we simply kill the existing service and start a new one with a new docker image — this means there is a little of “downtime” in terms of flow schedules and etc, like: • Time is 4.59 • We start a new deployment, meaning that at 4.59 the service is killed • There is a job scheduled to run at 5.00. • The service is only back up at 5.01. • This mean the 5.00 run will be fully missed. (please correct me if I’m wrong) My question, however, is about a second scenario, where we: • start run at 4.58, run takes about 5 minutes to finish, • We start new deployment at 4.59 — meaning there is a run in-progress • What happens with the flow run, if the entire service (local agent + code execution environment) is killed? • at 5.01 when the new deployment finishes, will prefect know to resume that flow run? how would it work? Reason I’m asking is because I plan on changing our deployment strategy to be blue/green, but I’m not sure how prefect will cope with a flow run being killed midway and etc Sorry if this is confusing! I appreciate any help.

Hey everyone I'm confused with the meanings of various flow runs - Failed, Cancelled, Finished, Skipped. Since I am migrating from Aiflow where we had the concept of "pausing" and "unpausing" a DAG, I'd like to know how we can replicate that behaviour over here. Ideally when I hit pause, all I want to do is stop rescheduling more flows until I hit "unpause". Although the tasks in the current flow should continue to run. How could I achieve that with Prefect ? (By the way I'm loving prefect with versioning, parameters, prefect cloud ~ all that 🤟)

Can I get answer to this problem related to build docker image please? I have put it in Stackoverflow, hoping to help with the others to search. When using docker as storage, in the call as below:

prefect register --project yourproject -p yourflow.py

It seems that perfect creates a

tmp{randomNumber}

folder and puts healthcheck.py and a Dockerfile, which participate the docker build. But I accidentally deleted the

tmp

folder, even after I recovered it, it still throws the following error. What happened was that Prefect expected a

tmp

folder with a new random_number as the suffix. How can I recover from it? https://stackoverflow.com/questions/70198040/error-in-prefect-when-build-docker-image-what-triggers-prefect-to-create-the-tm

Hello Prefect community, I'm getting a strange

ClientError

when using Prefect Cloud for a mapped tasks. The inputs to each task are identical, but some mapped tasks are successful while a few raise

DeprecationWarning: Using 'method_whitelist' with Retry is deprecated and will be removed in v2.0. Use 'allowed_methods' instead

(see screenshot for trace). Has anybody experienced something similar? I'm using urllib3 v1.26.7 and prefect v0.15.9

Hi, i am trying to install prefect server on gke using helm. instead of doing this

helm repo add prefecthq <https://prefecthq.github.io/server/>

i have cloned the repo

<https://github.com/PrefectHQ/server/tree/master/helm/prefect-server>

and so that i have this helm chart prefect-server locally and when i do helm install demo1 prefect-server i get the following error

Error: INSTALLATION FAILED: found in Chart.yaml, but missing in charts/ directory: postgresql

My question is regarding Storage. How does anything change if I add

flow.storage=Local()

to my Hello world flow ? Everything seems to be the same

Can someone help me with some graphQL stuff. I need to find all the flows that have a task running with a certain keyword.

Hi all! I have a question about the Schedulers… If I use two clocks, and both of them trigger at the same time, is my flow executed just once? Thank you!

Hey, Is there a way where I can get tasks results in flow state handlers?

I am trying

if new_state.is_successful():
        flow_run_id = prefect.context.flow_run_id
        flow_run_view = FlowRunView.from_flow_run_id(flow_run_id)
        b_task_run = flow_run_view.get_task_run("b-1")

but this wouldnt work when I run it locally, is there other trick?

Hey folks, we ran into some issues where task concurrency limits were being gobbled up by “ghost” tasks that were still marked running for Flows that were cancelled. We are able to retrieve those with the following GraphQL query:

{
  task_run(
    where: {_and: [{state: {_eq: "Running"}}, {flow_run: {state: {_neq: "Running"}}}]}
  ) {
    id
    start_time
  }
}

Maybe unrelated but for tasks which have a lot of runs, the dashboard seems to hit some kind of query limit and crashes when you try to see the list of Task Runs. The stuck ghost tasks don’t always have a lot of task runs related to them.