If I have a complex schedule with an hourly interval clock and a cron clock - will there be 2 runs 1 of the clocks take precedence for an overlap, or will there be 2 scheduled runs?

Hey all, has anybody successfully started a dask cluster + prefect on top of an existing spark cluster (databricks)? I have a databricks cluster but want to use distributed prefect to schedule

whats the correct pattern for spinning our own cloud runner? we simply want to add some initialization to

CloudFlowRunner

so that we can initialize sentry and datadog as early as possible, as well as flush our statsd buffer to track metrics. it seems like the place i'd expect it to read from the env var is hardcoded tho https://github.com/PrefectHQ/prefect/blob/master/src/prefect/agent/kubernetes/agent.py#L562-L586

Hi, does Prefect Cloud support multi-factor authentication? I’m looking through the features table at https://cloud.prefect.io/plans, and through my admin account profile page, and I don’t see a MFA option listed.

I am getting this error while executing a flow via Cloud Prefect 2.0 "RuntimeError: This portal is not running". Any clue what is going wrong. The same flow was working fine before I use Cloud Prefect 2.0. Actually it finishes all my flows and tasks successfully and at the end it crashes and it happens to all my existing flows. line 1512, in propose_state raise prefect.exceptions.Abort(response.details.reason) prefect.exceptions.Abort: This run has already terminated.

is there a way to find which tasks are holding the task tag concurrency locks? we’ve a case where our lock is taken up by a task when no flows are running

Fun with dataframes and map. So I have a map function that loads a dataframe in one task and then uses that dataframe in the next task. When I look at the return type on the first task, it is pandas.core.frame.DataFrame. When I look at the type of the dataframe value in the second task, it is class 'list'. I have the type hint set to pandas.DataFrame in the second task. Does anyone have any idea why the DataFrame is becoming a list?

Our prefect agent has been running without issue on ECS for the past ~11 months (since June 28 2021), but about 1 hour ago we noticed we were unable to submit new flows. Upon further investigation it appears our agent crashed and has been trying to restart but keeps crashing. I’m not sure if this is what initially caused it to crash but the error message we are seeing in cloudwatch logs now is :

ValueError: You have not set an API key for authentication.

Is anyone familiar with this error? Was something updated in prefect that could have caused this crash?

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.