When using the python

Deployment

object and specifying a remote S3 flow storage block and Kubernetes infrastructure, we are seeing strange behavior on flow execution. The deployment pushes the flow to S3 storage as expected (which is confirmed by the storage block in the Prefect Cloud UI being referenced in the deployment UI) but errs out with the following:

Flow could not be retrieved from deployment...FileNotFoundError: [Errno 2] No such file or directory: '/My/Local/Path/my_project/flow.py'

where the path is the absolute path of the machine that applied the deployment whereas the absolute path in the s3 bucket is just

<bucketname://flow.py>

. Here is the code we are using if anyone has any ideas?

from prefect.deployments import Deployment
from prefect.infrastructure import KubernetesJob
from prefect.filesystems import S3
from my_project.flow import entrypoint

infrastructure = KubernetesJob(namespace="prefect2")

deployment = Deployment.build_from_flow(
    flow=entrypoint,
    name="my_deployment",
    work_queue_name="default",
    storage=S3.load("default-block"),
    infrastructure=infrastructure,
)
deployment.apply()

Hi, has else had a problem with flow retries with deployments? I ran my flow directly in a python script and the retries were working fine. I then pushed the same flow to a deployment and then ran the deployment on an agent and the retries just didn't happen. (using prefect 2.1.1)

What replaces the following logic in prefect 2.0

schedule = IntervalSchedule(interval=timedelta(seconds=30))

with Flow("workflow", schedule=schedule) as workflow:

Is there a way to run a flow on an interval schedule using the

DaskTaskRunner

in prefect 2.0

Are there any plans (or existing methods) to allow users to save flow code to its storage block in an ad-hoc manner, independently of the deployment process? This seems like it could be nice for CI/CD workflows where only the flow code used by existing deployments is updated, but no new deployment actually needs to be created. It seems like the workaround for this right now is just to build a “dummy” deployment but not apply it, but I’m afraid that might introduce unintended side effects down the line.

Hi All. I have a dumb design question on how to use Prefect cloud + ECS Fargate. I have a flow registered to an ECS cluster. Now I want to iterate over a list of value for a parameter(e.g.[1,2,3,4,5]) that I used inside a task or the registered flow: example:

@task
def make_df(i):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hi from Prefect %s", prefect.__version__)
    <http://logger.info|logger.info>("this is the first step")
    data = {'Name':['Tom', 'Brad', 'Kyle', 'Jerry'],
        'Age':[20, i**2, 2*i, 18*i],
        'Height' : [6.1, 5.9, 6.0, 6.1]
        }
    df = pd.DataFrame(data)
    return df

How should I design my work to run 5 parallel task in ECS cluster?

See screenshots in thread! Hello! I am running into an issue when trying to build a flow on the Prefect CLI. Basically I am trying to use an S3 storage block, but when I run the command it just hangs and never finishes. The files do get uploaded to the specified S3 bucket, but no YAML file is ever created so I can't apply it to the deployment. Please let me know if you have a solution. Thanks!

Back again with DB QueuePool errors. But this time I have the whole stack trace! Anybody know how to solve this? Stack Trace in thread. I am running Postgres backend on Prefect 2.0

Hi. I wonder what the field name for the flow-run-name / unique-animal-names is. If I knew its field name this could help searching, also inside the code. Next question would be: How can I set it, I'd like to set my own silly-unique-names instead of these. Is there just a keyword parameter for it? And if there is no easy way, what is the hard way? Can I swap / hack / override some function or module? What is the module / function that creates the names. In my project it could be helpful to set this field to something more useful. Thank you!

Hi, I just signed up using my work email to try prefect https://app.prefect.cloud/ .But I didn't receive the verification email. I have been waiting since 2 hours. 😞

Hello, what is the correct way to create a deployment for a flow so that the flow code gets uploaded to remote storage? I have seen examples using

packager

or

storage

arguments but none of them are working for me.

deployment = Deployment.build_from_flow(
    name="docker-example",
    flow=my_docker_flow,
    packager=FilePackager(filesystem=RemoteFileSystem.load('minio-docker')),
    # storage_block=RemoteFileSystem.load('minio-docker'),
    infrastructure=DockerContainer(
        image = 'prefect-orion:2.1.1',
        image_pull_policy = 'IF_NOT_PRESENT',
        networks = ['prefect'],
        env = {
            "USE_SSL": False,
            "AWS_ACCESS_KEY_ID": "blablabla",
            "AWS_SECRET_ACCESS_KEY": "blablabla",
            "ENDPOINT_URL": '<http://minio:9000>',
        }
    ),
)
deployment.apply()

With the above code the deployment is created but the flow is not uploaded to minio

Hello everyone, I have a question about how to handle failures in flow runs: I have a basic ETL pipeline set up as a flow that is scheduled to execute daily, and it ingests data from about a 100 different endpoints. It is expected that the ingest for most of those endpoints will be successful, but also that there will always be several that are unsuccessful. But since some are unsuccessful, Prefect shows the entire flow run as a failure. Is there a way to change that? I'd prefer that it only show as a failure if none of the ingests are successful, because it's set up to alert people who use this about failures. The Flow object in Python is set up sort of like this

source_ids = ["source_1", "source_2", "source_3"]
with Flow("my_flow") as flow:
    for source_id in source_ids:
        data = extract(source_id)
        data = transform(data)
        load(data)
    return flow

Thanks in advance!

For Prefect 2: Is there a recommended way/best practice to generate logs from within one of my own modules? The modules are not contained in their own packages, they're just .py files in a folder (everything that's packaged seems to log properly). The modules are made up of some pretty big classes and functions and ideally these modules should not have any prefect dependencies. Unless there's no other way, then we'll have to live with it. 🙂 I'm probably missing something obvious, I don't have that much experience with the various ways you can set up Python loggers. I couldn't get this working using PREFECT_LOGGING_EXTRA_LOGGERS, but perhaps I've just set up my modules incorrectly.

(Prefect 2) I want my flow to write a file (more technically, I want one of the tasks to do it). My file system block is GCS but its preferable if it outputs locally. Currently, the file seems like it isn't created at all. What to do?

Hi 👋👋👋 Is there a command in a Prefect flow that will stop (kill) the current agent? To give some context, our agent is a cron job in Render (Heroku alternative) that wakes up for time to time and starts the agent to process flows. But we don't need it to run all the time, so after it got a flow from a queue and flow is done, we want the agent to stop (and then the cron job in Render will also stop). I think process.kill in the end of the flow might work but seems a bit wrong.

Hi there, is there a guide or steps to setup prefect 2 environment to run orion server and agent in different containers in different servers?I was able to do that but the agent refuses to connect to server when using https (it only works with http) is there additional configurations needed to have agent connected to orion server over https?

Hi all, I have a question about Orion, is there a way to call tasks from other tasks? In prefect 1.0 it's possible to call

some_task.run(args)

from other tasks and it's very handy, cause very often it's necessary to call existing tasks (especially from prefect task library) deep inside other tasks. in Orion I didn't find how to do that, but pretty sure there should be some way?

Hi All, new to Prefect ... would anyone be willing to link some open source projects that use Prefect? Basically I just want to see some more complex scenarios than what is shown in the docs. Like a more complicated graph dealing with many different tasks and states. Thanks for your help!

Hi All, i have orion up and running connected to postgresSQL database,it was working fine until i shut down the orion and tried to run it again,orion in trying to create the database tables again! as shown in the image below is there a way to avoid initializing database tables on every orion start up?

Hi All, new to Prefect. I found that we can put the schedule parameter under prefect.deployments in Python API May I ask what kind of schedule we can put? Thanks.

Hello All, Is there a way to secure Orion UI? I am not using prefect cloud so I am looking for a way to have authentication on Orion UI instead of being publicly accessible

Is it possible to stick with prefect 1 or in the nearest future it would be deprecated for prefect 2?

Hey team - I had a flow run kick off unexpectedly (Cron

10 15 * * 1-5

and kicked off on a Saturday). How can I debug this? More details inside

Hey - prefect 1.0 question. Is there a way for me to do a

try except

over my entire flow ( many tasks included ) so that I can log this to a third party ? I know I can send slack notifications through the automations but I am looking for something more granular where I can share metadata about the flow and not have to query the GraphQlApi. I would like this to be part of the flow itself.

Hi, I am using graphql api to query prefect database but getting timeout exception.

requests.exceptions.ReadTimeout: HTTPConnectionPool(host='localhost', port=4200): Read timed out. (read timeout=15)

How can I change or configure this timeout ? Also we're running this query for our locally hosted prefect (prefect 1.0)

def readDatabase(query: dict) -> None:
    client = prefect.Client()
    data = client.graphql(query)

Hey, I want to create a dynamic DAG (where flows are not directly written into the code using the @flow decorator) However, I need to set dependencies for these subflows so a proper DAG is created. But Prefect 2.0 doesn't seem to have any argument for setting the upstream dependencies for flows. What could be a possible workaround for this?

Hi! I want to pass a custom variable for my state handler to use it, how can i do that?

The other day I got some advice to look at extra_loggers for handling logs from my packages, which worked great (thanks Anna)! Then I found logging.yml, which was even better. I had some issues getting Prefect to actually respect logging.yml when I finally deployed everything, and it appears that the problem is specifically with docker infra runs. If I set up an agent for local/process runs it's all good. Is anyone aware of a workaround for this issue, for example, can I put logging.yml somewhere in my docker images to have it get picked up?

Hey all, does Prefect 2 have flow concurrency limits? I’ve read about work queues but the description on Work Queue Concurrency isn’t particularly clear:

Each work queue can optionally restrict concurrent runs of matching flows.

Does this mean that if I have

Flow_A

and

Flow_B

running, they can run concurrently for any concurrency limit value?

Hello, I have a question about prefect 1 (1.3). We are currently running a very outdated instance on our own server (0.15.9) and we are in progress of getting that to the cloud. We are now first trying with version 1.3. But when trying to run a flow I get this error: Failed to load and execute Flow’s environment: FlowStorageError(’An error occurred while unpickling the flow:\n AttributeError(“Can\’t get attribute \‘_make_function\’ on <module \‘cloudpickle.cloudpickle\’ from \‘/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle.py\’>“)\nThis may be due to one of the following version mismatches between the flow build and execution environments:\n - cloudpickle: (flow built with \‘2.1.0\’, currently running with \‘2.0.0\’)\n - prefect: (flow built with \‘1.3.0\’, currently running with \‘0.15.9\’)\n - python: (flow built with \‘3.10.4\’, currently running with \‘3.8.12\‘)’) I have created it with prefect 1.3 on my mac, so that should work right? But it still seems to somehow mix the two old version. Any idea of where I’m going wrong?

I have the following flow:

@flow
def test():
    print("hello world")

And it will fail retro-actively from time to time:

15:18:12.953 | INFO    | prefect.agent - Submitting flow run '8b6a74ba-6dbe-49bd-b303-d5e9c863a087'
15:18:13.097 | INFO    | prefect.infrastructure.process - Opening process 'dangerous-ostrich'...
15:18:13.104 | INFO    | prefect.agent - Completed submission of flow run '8b6a74ba-6dbe-49bd-b303-d5e9c863a087'
15:18:30.922 | ERROR   | Flow run 'dangerous-ostrich' - Crash detected! Execution was interrupted by an unexpected exception.
15:18:30.954 | INFO    | prefect.engine - Engine execution of flow run '8b6a74ba-6dbe-49bd-b303-d5e9c863a087' aborted by orchestrator: This run has already terminated.
hello world
15:18:35.274 | INFO    | prefect.infrastructure.process - Process 'dangerous-ostrich' exited cleanly.

As you can see above, it completed successfully but at the same time it didn’t. Furthermore, the code clearly ran (

hello world

was printed), but it was still marked as an error. Shouldn’t the flow just be marked as completed?