Hey Prefect community, I'm working on deploying an instance of an Orion server to ECS using AWS CDK. One issue I've run into is how the database url is being specified via env vars. In ECS, there is a distinction between environment_vars and secrets and, to my knowledge, they cannot be concatenated. This presents a dilemma as you know must: • Have the database password in the clear as an env var. ◦ This also opens up an issue, where if the password is updated, it is not reflected until the next deployment. • resolve the secret key, concat with the whole URI, put into SSM

orionServer.addContainer('ServiceContainer', {
            essential: true,
            image: ContainerImage.fromDockerImageAsset(serviceImage),
            logging: new AwsLogDriver({
                streamPrefix: 'orion-server'
            }),
            portMappings: [{
                containerPort: this.props.orionEnvVars?.port || Number(this.CONTAINER_PORT)
            }],
            environment: {
                PREFECT_ORION_API_HOST: this.props.orionEnvVars?.host || '0.0.0.0',
                PREFECT_ORION_API_PORT: this.props.orionEnvVars?.port?.toString() || this.CONTAINER_PORT,
                PREFECT_ORION_DATABASE_CONNECTION_URL: `postgres+asyncpg:///${this.props.username}:${ecsSecret.fromSecretsManager(this.props.dbPassword)}@${this.props.databaseHost}/orion`

            },
            secrets: {

            },
            memoryLimitMiB: 300,
        });

We can overcome this by altering the settings provided, and fetching the username and password separately from the host to build the DB URI, perhaps as a second set of options. Has anyone found an alternate way to concat the strings together? If not, are we open to adding additional options to distinguish DB_PASSWORD to be injected as secrets?

This message was deleted.

Hello everyone, I’m working on running a very simple Prefect 2.0 workflow using a remote Ray cluster. More details in the thread below

friends, I’m trying to setup prefect on my k8s cluster using the official helm chart. On top of that I want to enable hasura admin secrets. It seems like when I’m doing it the graphql service is not starting. Correct me if I’m wrong but looks like

database upgrade

command does not respect

PREFECT_SERVER__HASURA__ADMIN_SECRET

value.

Hi there everyone, I’m trying to understand how to do simple map tasks with prefect2

@task
def generate_numbers():
    return [1, 2, 3, 4]

@task 
def compute_sth_expensive(number):
    return number ** 2

@flow
def pipeline():
    result_generate_numbers = generate_numbers()
    results = map(compute_sth_expensive, result_generate_numbers)
    for r in results: r.result() ## ??

Is that an acceptable pattern? I want to do a parallel calculation over the

result_generate_numbers

and then perform some gather-like operation

hey community, i’m switching to Prefect 2.0 cloud now and trying to programmatically register a deployment in the cloud. I have a webhook that when called need to register a deployment in the deployment cloud. What do I need to do to achieve this? Right now I have a very simple piece of code that fails because it calls into sqlite3 which is unexpected. I thought that Prefect will just call api cloud w/o any databases. How do I configure the process so that it starts calling the cloud api?

import flow
from prefect.deployments import DeploymentSpec
from prefect.orion.schemas.schedules import CronSchedule


def create_deployment(name: str,
                      user_id: str,
                      job_id: str,
                      schedule: str):
    d = DeploymentSpec(
        flow=flow.flow,
        name=name,
        schedule=CronSchedule(
            cron=schedule
        ),
        parameters={
            user_id: user_id,
            job_id: job_id
        }
    )

    d.create_deployment()

Hi, What is the best approach to be able to run flow locally not changing Flow() parameters? don't want to re-invent the wheel. way it should be deployed and kept in code:

with Flow(name="Flow", executor=executor, storage=storage, run_config=run_config) as flow:

way it can start locally:

with Flow(name="Flow") as local_flow:

flow itself remains the same.

Hi, I've set up a Postgres Orion database for registering the states of flows/tasks etc. I noticed they create a structure in public schema. Is it possible to have it create the structure in say prefect-Orion? Use case is that I may want to set up logical replication from that database to another and I would prefer to avoid public (it's already in use by some other database. )

Hi! How to migrate from Prefect cloud 1.0 to 2.0?

Hi! I’m just a beginner with prefect. I was trying to run the agent on the ec2 t2.micro instance, but an out of memory error occurred while running it, so I think it is an insufficient memory problem, so I am trying to upgrade the ec2 instance. Are there any recommended specs for running the agent?

Morning, I'm trying to register flow from another python (like it's described in https://github.com/PrefectHQ/prefect/discussions/4042) My code which does registration:

import os.path as p

from <http://flows.xxx|flows.xxx> import testflow
from prefect.storage import S3

flows = [testflow]

root = p.dirname(p.realpath(__file__))

storage = S3(stored_as_script=True, key='testflow.py', bucket='test')

if __name__ == '__main__':

    for flow_file in flows:
        flow = flow_file.flow

        print(f"Registering flow {flow.name} from {flow_file}")

        storage.add_flow(flow)

        flow.register(
            project_name='test',
            idempotency_key=flow.serialized_hash()
        )

Flow itself:

name = "testflow"
executor = LocalDaskExecutor()
storage = S3(stored_as_script=True, key='testflow.py', bucket='test')
run_config = KubernetesRun(
    job_template_path='<https://XXX/job_template/k8s_job_template.yaml>')

with Flow(name=name, executor=executor, storage=storage, run_config=run_config) as flow:
	....

Error I've got is the same when you're not adding storage to DataFlow:

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

What am I missing?

Hello everyone! Is Orion ready to be used in PROD? We are coding some new pipelines and we want to start using it from the beginning instead of upgrading in the future. Thank you in advance!

Hey guys, i have a quick question about conditions in flow definition. I use a boolean parameter and build the flow depending on its value. I'm planning to use a prefect case to do so where i will define new variables from tasks call. Are those definition globally scoped so i can use them "outside" of the prefect case or are they locally scoped ?

Hi all, I have an authentication problem:

prefect.exceptions.AuthorizationError: Malformed response received from Cloud - please ensure that you are authenticated. See `prefect auth login --help`.

However when I run

prefect auth status

it says I'm connected and authenticated. Any idea what's wrong?

Hello, is there a simple way in OSS Prefect v1 to set container images at the task level when using kubernetes executor? From my understanding the answer is no, but could be wrong. An example is a flow containing N tasks of native type, e.g. ShellTask/PythonTask, each relying on a (possibly) separate container image.

how does Prefect 2.0 deal with dependencies? Does agent need all deps that flow requires? Also, what if the flow file depends on local files? I.e. Here is my

flow

directory:

(venv) ➜  src git:(main) ✗ ls -la flow                                 
drwxr-xr-x  6 avysotsky  staff   192 May  9 08:47 .
drwxr-xr-x  5 avysotsky  staff   160 May  9 08:34 ..
-rw-r--r--  1 avysotsky  staff  3301 May  9 08:47 flow.py
-rw-r--r--  1 avysotsky  staff   870 May  8 10:11 graphql.py
-rw-r--r--  1 avysotsky  staff   611 May  9 08:47 prefect_client.py

And here is how I create the deployment:

d = DeploymentSpec(
        flow_location="./flow/flow.py",
        name=name,
        schedule=CronSchedule(
            cron=schedule
        ),
        tags=[
            f"user_id:{user_id}",
            f"job_id:{job_id}"
        ],
        parameters={
            user_id: user_id,
            job_id: job_id
        }
    )

Is Prefect smart enough to pull in all the deps that flow.py needs?

Hi, I am trying to set upstream tasks for a given task dynamically. It is working fine however I do see some extra tasks listed up which don't do anything (screenshot -1). Why is that happening? When I am printing set_upstream tasks in console, it is showing just the tasks I want, however while executing, I see more (screenshot-2).

Hi everyone, I was wondering if anyone knew a way to set a

resource_manager

cleanup

method upstream of another task?

Is there documenting on using attachments with the https://docs.prefect.io/api/latest/tasks/notifications.html#emailtask that I'm missing? I'd like to use the result of one task, a Pandas DF, and attach df.to_csv(..name..) to the email. It asks for the "names of files" but are these PathLikes

In 2.0, is there an equivalent to modulestorage which works for k8s and docker flow runners?

Dear Prefect-community, We use Prefect Orion 2.0 with Cloud Prefect 2.0 UI and while launching beta.prefect.io with a IFrame for our Application we get the following error message: "*beta.prefect.io* refused to connect". Looks like in the Cloud Prefect setup we need to add x-frame-options:SAMEORIGIN. Since I don't have control I don't know how to resolve this issue.

Hey does this just mean our EKS cluster hasn’t whitelisted api.prefect.io?

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