what's best practice for dealing with Storage? We have our flows in github and for testing we want them to run locally in checked out repo using

Local

Storage, but then for production we want the same flow to run on our ECS agent using a

GitHub

storage pointing to the same repo. With the

UniversalRun

config it can run on either agent, but how do we make the storage conditional so it uses the correct storage depending on which agent it's targetted to?

hello again! We noticed some K8s prefect jobs running for a suspiciously really long time (48 days in some cases). After looking at the logs; some of them seem to be stuck on

[2022-06-24 09:26:07+0000] INFO - prefect.S3 | Flow successfully downloaded. ETag: "032fd4a15a8c777534d795042e1c0ef6", LastModified: 2022-06-23T15:24:59+00:00, VersionId: None

To my understanding, this is the logging from Prefect when the flow is being downloaded from s3 (before running any tasks). We don’t see these flow runs in the UI either. How do we troubleshoot this and prevent this from happening in the future. Shouldn’t zombie killer have ended these flow runs? If they still had a heartbeat shouldn’t they show up in the cloud UI?

v2.0b7 :: OK, I feel stupid asking this. How do I get the current profile within a task?

I want to run the same DbtShellTask on two different schemas in one flow. Is there an efficient way to do this? I am currently doing this and it feels chunky and repetitive:

code snippet moved to thread

The two DbtShellTasks are identical except for the schema. I tried defining the DbtShellTask without schema outside of my flow and tried a few different things inside my flow:

transform_task.run(dbt_kwargs = {"schema": "schema_1"})

but I always get the same error:

ValueError: Could not infer an active Flow context while creating edge to <Task: DbtShellTask>. This often means you called a task outside a `with Flow(...)` block. If you're trying to run this task outside of a Flow context, you need to call `DbtShellTask(...).run(...)`

I can tell I am doing something silly but can't seem to resolve it 😅

Hi, can someone please explain what are proprietary extensions within the Prefect environment? Any examples?

@Kevin Kho hi kevin， is there any plan to support Celery Runner? at most map scenario, we just want a distribute queue to run many task and a rich web ui to manage these task in a batch. so sometime we don't need a complicated engine like dask to do this, because some time it will make some confused problem( e.g., run a task twice )

Hi. I’m trying Prefect Orion. I tried to start Prefect Server using the Prefect Python API, but got an error. Anyone have any idea what the cause might be? I would be glad to know. I’m using Prefect 2.0b7.

from prefect.cli import orion

# Start Orion server.
asyncio.run(start_orion_server())

async def start_orion_server() -> None:
    await orion.start()

    print("Succeed starting Prefect Orion Server.")

    return

Hi , We have installed python 3.8 version in jupyter notebook and installed the prefect 1.2.2 version. However, during the prefect flow registration, we are facing the below issue. In the prefect 1.2.2 changelog documentation, we found that it supports python 3.10. Can you please tell if the below error is because of the incompatibility(if any) between python3.8 and prefect 1.2.2 versions.

Unexpected error: ClientError('400 Client Error: Bad Request for url: https://api.prefect.io/graphql\n\nThe following error messages were provided by the GraphQL server:\n\n INTERNAL_SERVER_ERROR: Variable "$name" of non-null type "String!" must not be\n null.\n\nThe GraphQL query was:\n\n query($name: String!) {\n secret_value(name: $name)\n }\n\nThe passed variables were:\n\n {"name": null}\n') Any clues?

👋 When I schedule a Flow every "12 hours" using the

IntervalSchedule

object. Would the flows start at different time or all at the same time ? What I am looking for is to distribute the load on our servers when flows run. And so a new flow should start at a random time and then follow the

IntervalSchedule

constraint.

Is anyone else seeing jobs fail to start this morning using the hybrid hosting model? Our jobs are not getting picked up this morning but we have not seen any errors in the logs on our Agents running in our k8s cluster and did not make any changes for our k8s environment or our Agents

Is there a way to rename pipelines? Or "retire" them? We have the old

data_pipeline_prod

that we want to somehow label that it is deprecated? And then change the one with test in its name to the new

data_pipeline_prod

? Or any suggested ways to do this?

I want to speed up CI/CD by self-hosting prefect server and having the CI tests register flows against a our self-hosted server. How can I direct the flow.register to point to our custom URL?

How does one add a new Parameter to an existing Flow? I inherited a Flow from a colleague that has since left and I wish to add a new Parameter

with Flow("HathiTrust Deposit (MARCXML)") as flow:    
     deposit = Parameter("deposit", default=False)

This is the existing parameter. I want to add a second so naturally I tried adding another underneath the first like this:

with Flow("HathiTrust Deposit (MARCXML)") as flow:  
    deposit = Parameter("deposit", default=False)
    local_test = Parameter("local_test", default=False)

This results in an error:

UserWarning: Tasks were created but not added to the flow: {<Parameter: local_test>}. This can occur when `Task` classes, including `Parameters`, are instantiated inside a `with flow

How does one add a parameter to an existing flow?

Is github storage no longer an option for prefect 2.0?

Hi there, I have a question about iterated task mapping...

can we do resource managment in prefect ? giving different flows differnt cpu power

is it worth renaming this channel 1.0 and having a dedicated 2 channel, sure gets confusing

Hi, question regarding flow registration - During follow registration, it requires packages used in the flow to be already installed on the machine. One walkaround is to import packages inside each tasks, however, is there a way to not duplicate code and import at top level? Thank you!

Hi! Small question about Prefect 2.0. Is there any concept similar to Secrets in Prefect 1.0? I used to store all my secrets in

config.toml

and access them via Secrets API or

prefect.context.secrets.get()

. But I can't find anything similar in 2.0

When a Flow loses the heartbeat and then is restarted (No heartbeat detected from the remote task; retrying the run.This will be retry 2 of 3.) is there a flow State provided when the process starts again? Like a Retrying state or a Running state?

Hi, having a complete brainfart moment. I am trying to return the string “full”, to the next task, but I am getting None:

@task(nout=1)
def determine_run(clock):
    if clock == "full_run":
        <http://logger.info|logger.info>("full_run clock")
        run_type = "full"
        <http://logger.info|logger.info>(f"{run_type}")
    return run_type

Hello All, I am running prefect orion in docker with sqlite3 DB. For persisting data used bind mount. The location on host cannot be local folder as after weekly boot New host will be assigned and docker will run on that. Hence I used AFS like my other bind mounts, however having such setup and on prefect orion start its stuck at message "server starting". If I set db directory just inside docker then server gets started but the data cannot persist now. Is there any issue keeping db file on AFS? What can be done in this case?

I'm on v2.0b7 and am trying to get a minimal code example to work with a Dask cluster. I can use DaskTaskRunner locally, but as soon as I add the TCP address to my test cluster, the system fails. I see a very lengthy error on the Dask worker that includes: "prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url" I am able to submit jobs successfully using Dask directly, just not through Prefect. Here is the code:

from prefect import flow, task
from prefect_dask.task_runners import DaskTaskRunner

@task
def say_hello(name):
    print(f"hello {name}")

@flow(task_runner=DaskTaskRunner(address='tcp://[redacted]:8786'))
def greetings():
    say_hello('test')

if __name__ == "__main__":
    greetings()

Any ideas?

In Prefect 2.0 (Orion), can we start the agent with a custom job template like how we could do in Prefect 1.0? https://docs.prefect.io/orchestration/agents/kubernetes.html#custom-job-template

Hi~ Can i use the company account for register to Prefect Cloud? I sign up the Prefect Cloud by using Google groups mail but can't find the activation mail from okta. Also Instruction mail about reset password can't recive to mail. 😓

Hi, I'm quite looking forward to all the features coming up in prefect. I was wondering however if there is some roadmap somewhere about next steps/features that I should expect.

Hi All, When we are trying to register the prefect flow , we are facing the below error. We are using prefect 1.2.2 and python 3.8. The prefect server is installed on Google kubernetes engine. pls check and suggest.

In cloud 2.0 whats the best strategy to install a k8s flowrunner dependency - do I patch a command in to pip install?