Hi, a bit new here so forgive me if this is a bit rudimentary, but I was reading https://www.prefect.io/blog/introducing-prefect-2-0/ and the following stuck out: "_Prefect 2.0 introduces an ephemeral API that lets it deliver a full orchestration suite even when you haven’t spun up an Orion server, database, agent, or UI._" I am a bit confused how that works internally and have a couple questions if you don't mind: 1. If I execute a flow locally in process (e.g.

___main___

), is there an in-process Orion orchestrator running alongside it? 2. If 1), is the flow meta-data persisted in a local SQLite instance on disk? 3. Is this "ephemeral" server-less workflow intended for purely testing purposes, or is it production viable?

Hi, I'm new to Prefect and trying out some basic stuff, and got a doubt. I have three tasks

a()

b()

c()

each of which waits for 2 secs and prints the current time. Next, I created the following flows:

with Flow("flow-a") as flow_a:
        a()
        b()

    with Flow("parent-flow") as flow:
        c()
        flow_a.run(executor=LocalDaskExecutor())
    
    flow.run(executor=LocalDaskExecutor())

Here, to my surprise, task

a

and

b

run first, then after 2 seconds task

c

runs. Example o/p:

b: 23:04:32
a: 23:04:32
c: 23:04:34

I want all 3 to run parallelly, what am I doing wrong? (Sorry, if this is not the right forum to ask these questions.)

hi there, prefect2 question: are there any examples for processing a database with 1bln rows? I’m suspicious of tasks doing row-by-row operations and performance. Likely a better choice is pagination, with a generator task yielding successive pages, that are then processed in parallel by another task? How can that be expressed in prefect2?

hey there! using S3 Storage Class, how do I modify the path of the results?

In Prefect 2.0, are the

S3Download

,

S3List

, and

S3Upload

tasks (available in 1.2 from the

prefect.tasks.aws.s3

module) available?

I have a flow that runs locally but is returning a None result when run in Prefect Cloud. I think it's because I'm not setting the S3Result as the default? But this is confusing since I'm using the S3Storage as the default for flow storage. Does this seem right?

hi! is there a way to get the image running the job in kubernetes? so after a flow is triggered, can I have access to the image name running the container?

had a question about prefect(1.0 in this case): it isn’t super clear from reading the documentation where different aspects of the system “live”… it seems like, assuming Prefect Cloud usage, you have the cloud, and then some agents. However the documentation jumps into showing CLI-esque commands and doesn’t provide much clarity around where that lives

When should we expect a prefect docker image based on python 3.10+?

i have a general question about how best to utilized Prefect for training (and Pickling) machine learning models. This is my mini-MLE setup: I have a Gitlab repo that holds all my model definitions. (model-repo) I have a Giltab repo where Prefect lives. (prefect-repo) I use poetry to create a package with all my model definitions. I import that into the Prefect repo, and use prefect's tasks to train, pickle the models. This is great, but super super slow and a terrible development process. Every time I make a change to a model in the model-repo, I have to recreate / repackage the Repo. I have to get changes in to main, update a tag version, etc etc. Then I have to update the pyproject file in the prefect-repo (ugh). There's got to be a better way. Curious how others do MLE with Prefect. Thanks! 🙏

Is there a way to unit test tasks where

.fn

returns a coroutine? I tried this:

from prefect_aws.s3 import s3_upload

...

with prefect_test_harness():

my_upload = s3_upload.fn(bucket=S3_BUCKET_NAME, key='B5_key', data=b'55555', aws_credentials = AwsCredentials())

asyncio.run(my_upload)

But it’s not working with this runtime error:

E           RuntimeError: There is no active flow or task run context.

../valo-prefect-poc/.venv/lib/python3.7/site-packages/prefect/logging/loggers.py:91: RuntimeError

Hello all, I am trying to install prefect 2.0

pip install -U "prefect==2.0b1"

and I am encountering below error my machine

Ubuntu 20.04.4 LTS

.

The basic tutorials (https://orion-docs.prefect.io/tutorials/first-steps/) works even with the error but I am facing another issue while running a simple POC Regression model with prefect. Just want to make sure if the issue is not due to the above error.

flow.register is giving me this error

Newbie ALERT: Trying to run a basic regression model using

prefect 2.0

. Facing below error:

Traceback (most recent call last):
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/prefect/engine.py", line 467, in orchestrate_flow_run
    result = await run_sync_in_worker_thread(flow_call)
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/prefect/utilities/asyncio.py", line 52, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(context.run, call, cancellable=True)
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/anyio/to_thread.py", line 28, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(func, *args, cancellable=cancellable,
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 818, in run_sync_in_worker_thread
    return await future
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 754, in run
    result = context.run(func, *args)
  File "car_linearregression.py", line 104, in do_regression
    X,y = get_feat_and_target(df_car,target)
TypeError: cannot unpack non-iterable PrefectFuture object

It's a simple serial execution where a

flow

function calls different

Task

functions serially. very similar to the below tutorial on the official website.

import requests
from prefect import flow, task

@task
def call_api(url):
    response = requests.get(url)
    print(response.status_code)
    return response.json()

@task
def parse_fact(response):
    print(response["fact"])
    return 

@flow
def api_flow(url):
    fact_json = call_api(url)
    parse_fact(fact_json)
    return

So far I have tried creating a new virtual env and install minimal packages required to run the ML model but had no luck. Could you please help me with this.

Seeing a strange error when scaling up a prefect workflow

File "/venv/lib/python3.8/site-packages/prefect/orion/orchestration/rules.py", line 534, in __aexit__
    await self.after_transition(*exit_context)
  File "/venv/lib/python3.8/site-packages/prefect/orion/database/dependencies.py", line 112, in async_wrapper
    return await fn(*args, **kwargs)
  File "/venv/lib/python3.8/site-packages/prefect/orion/orchestration/core_policy.py", line 190, in after_transition
    cache_key = validated_state.state_details.cache_key
AttributeError: 'NoneType' object has no attribute 'state_details'

Any idea how to debug/interpret this?

Hey Prefect, long time! I'm getting some buggy flows stuck running for 24 hrs + that should be done in 10 minutes approx. Is there any way I can set a maximum run time for my flows? I use an ECSRun config. Cheers.

Hello everyone! How should I manage Secrets on Prefect 2.0? I cant find information on the orion-docs website 😞

Hi, we are using prefect Local Agent wrapped in a docker container, we would like to switch to Docker Agent. Does someone knows if it's possible to setup a Docker Agent with Docker IN Docker (dind) ? Thanks !

Hi, How to install prefect CLI on mac ?

Has anybody successfully installed Prefect on a GCP Compute Engine instance? I'm following this tutorial but running into some issues during the installation: https://medium.com/the-prefect-blog/prefect-server-101-deploying-to-google-cloud-platform-47354b16afe2

ModuleNotFoundError: No module named 'markupsafe'

Hi Prefect team. I am trying to use the great Expectation task in my prefect flow and i cannot mange to get it work with custom expectations that i wrote myself - https://docs.prefect.io/api/latest/tasks/great_expectations.html#rungreatexpectationsvalidation. It doesnt seem to pick them up. Would you know if we can use custom expectations with prefect, and if we can, does it require any additional import in the code? Thank you for your help !

I maintain a tool to package (set

flow.storage

and

flow.run_config

) and registration flows for my enterprise. Recently we wanted to duplicate all Prefect Cloud UI logging to Cloudwatch. Inside an individual flow I can add this code to get the logs there, but I was wondering if there was a way I could do this through a central utility?

with Flow("My First Flow") as flow:
    logger = context.get("logger")
    logger.addHandler(
        watchtower.CloudWatchLogHandler(
            log_group_name="prefect-logs",
        )
    )

Aka, would something like this work?

from prefect.utilities.storage import extract_flow_from_file

flow = extract_flow_from_file("path")
flow.logger.addHandler()?
. . .
flow.register()

Managed to run my prefect2 flow on an agent and consuming from a work queue, with no problem! Not bad, #PrefectJoys When can we expect multiple workspaces and team-workspaces? (if ever?)

I'm getting an error attempting to schedule a flow. The "Schedule" toggle, in this case, is on and was turned on without error. The GraphQL error only pops up after submitting the schedule

Hi again! I’ve got a Prefect2.0

prefect-gcp

authentication issue/question. I’m trying to follow the example docs, but getting an error Importing

GcpCredentials

:

NameError: name 'SecretManagerServiceClient' is not defined

. FIY the prefect1.0 GCP/bigquery tasks are working fine in my environment. Any advise?

Hey all, I'm working on a Prefect 2.0 implementation on docker running on GCP compute engine. One issue is the bash script to get everything up and running. This is what I came up with, which seems to work. I'm wondering if I am overcomplicating this and there is an easier way to do it? The same thing was much simpler with Prefect 1.0 (set api key, register flow, then start agent).

prefect cloud login --key <MY_KEY> -w <MY_WORKSPACE>

rand="GCS_"$(cat /dev/urandom | tr -cd 'a-f0-9' | head -c 16)

output=$(printf '%s\n' 2 <MY_BUCKET> <MY_PROJECT> $rand | prefect storage create)

storage_id=$(echo $output | grep -oP "(?<=identifier \').+?(?=\')")

prefect storage set-default $storage_id

prefect deployment create my_flow.py

output=$(prefect work-queue create my_queue | grep -oP "(?<=UUID\(\').+?(?=\'\))")

prefect agent start $output

follow up

prefect-gcp

issue using the

bigquery_insert_stream

task. I’m not able to pass

Timestamp

data-types -getting:

TypeError: Object of type Timestamp is not JSON serializable

Hi again! I am using

create_flow_run

and

wait_for_flow_run

(see previous thread). The child flow takes a few hours to run, and in the meantime, the parent flow decides that it must have failed and reports that the whole parent flow failed. Is there a way to extend the timeout period for

wait_for_flow_run

?