Hi everyone, having a bit of a configuration problem. I have a script that runs fine when I run it locally but when I register it with Cloud, it fails to execute on there with the following error:

Failed to load and execute Flow's environment: ValueError('No flows found in file.')

This is the flow definition:

if __name__ == '__main__':

with Flow('Scouter-Solr-Script') as flow:

snowflake_user = PrefectSecret('snowflake_usr')

snowflake_pwd = PrefectSecret('snowflake_pwd')

snowflake_to_solr(snowflake_user, snowflake_pwd)

flow.run_config = LocalRun(labels=['SVRNAME1'])

flow.storage = Git(repo="Prefect-Flows", flow_path="Python/Scouter/snowflake_to_solr.py", git_clone_url_secret_name="azure_devops_clone_url")

flow.register(project_name='Scouter')

I'm assuming it has to do with the

if __name__ == '__main__'

part on top of the Flow class definition, but I don't know how to go about fixing this. I need that for multiprocessing that happens within the main task of the flow.

hi everyone, is it possible to add annotations to k8s jobs created by 2.0 agents? currently we inject Vault secrets via annotations and it works fine in 1.0 (we have a custom job template deployed with the agent) but I’m trying to scope the migration process and I haven’t seen any equivalent in 2.0 (I also looked at the agent code for 2.0b6)

Hi all, relatively new to prefect 2.0 and using the cloud to get going. Had two questions that somewhat baffle me so far and I couldn't find in the docs: 1. regarding the storage. Configured remote S3 storage... what I'm not sure about is if this storage is used somehow in prefect? I had to give aws keys and I wasn't sure if this is somehow kept on prefect side (cloud). 2. i made a scheduled deployment, every 300 secs. when i look in the UI it seems to have put there about 100 flow runs into the future, but I can't figure out why approx 100... why not 20.. or 50.... I don't remember seeing any configuration for this.

Hello! where can I find out more info about what changed in the block api for beta6? I have a script that configures minio as default storage and I cant get it to work with the new apis. Before beta6 there were 2 abstractions bloc_spec and block and now there are 3? block_type, block_document and block_schema? thanks! https://github.com/fraibacas/prefect-orion/blob/main/flows/init_orion.py#L15

Custom Roles Is it possible to limit access of users to specific projects in their "workspace" using "Custom Roles" in the enterprise edition ? If yes, is there an example? I do not see it in https://docs.prefect.io/orchestration/rbac/overview.html

hey all — is there a straightforward way to run a flow continuously, sequentially and in perpetuity? like an infinite loop.

Is it just me, or is the UI broken for flow runs at the moment? https://cloud.prefect.io/zesty-ai/flow-run/fc3a37ca-bcce-44b9-ae05-572ecf91328d

Where can I find the signals FAIL, SKIP, etc. in Prefect 2.0? I am getting an error when importing them from the old place.

I'm attempting to use map to reuse tasks and am running into the following error. Any insight into what I might be doing wrong?

Unexpected error: KeyError(0)
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/flow_runner.py", line 569, in get_flow_run_state
    executors.prepare_upstream_states_for_mapping(
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 681, in prepare_upstream_states_for_mapping
    value = upstream_state.result[i]
KeyError: 0

Hey guys, i started getting this error when running my flows on ECS Agent:

An error occurred (ThrottlingException) when calling the RegisterTaskDefinition operation (reached max retries: 2): Rate exceeded

This means that i should increase my Rate limit on my AWS account?

Hi. is it possible in prefect 2.0 to delete all deployments at once? Thanks

Is there an ECS flow runner planned for 2.0? Couldnt find anything in the v2 tagged github issues... Thanks!

Getting this message.

TypeError: cannot unpack non-iterable PrefectFuture object

Do I just need to do

@task(nout=2)

in Prefect 2.0?

One last question for today... Mapping Tasks... is this still TODO on Prefect 2.0?

Is the Cloud RBAC granularity at the tenant level? I.e. I’d like to have everyone be a reader, but some users to only be able to “write” some flows but not others. Separately, I’d also like to have “per-team” agents and some way to limit which agents can a user submit flows to. Am I better sticking to just separate tenants/separate servers for on-prem deployment, than considering the RBAC for this? Is

workspace

the 2.0 cloud analogue to the 1.0 tenant?

Hi, someone know if there is a way of installing just the core of Prefect Orion, without the REST API, UI, CLI, just the python API? Thanks in advance.

Is there a way to force a specific mapped task to run in sequence even though the flow is using a local dask executor? I have a flow with a bunch of mapped tasks that I want to run in parallel but one mapped task that i'd like run in sequence.

Hi folks, I'm trying to schedule a flow but keep getting

Something went wrong. Please wait a few moments and try again

. Flow runs fine locally. Any idea how to troubleshoot this?

Hi everyone. Has somebody successful ran flows that use Dask Dataframes or multiprocessing library on a Windows Local Agent run deployed on Cloud? I am finding it impossible to define a flow and wrap it inside the necessary

if __name__ == ‘__main__’

. It all works fine when running locally with

flow.run()

but it tells me it can't find a Flow when I deploy it to Cloud and run it from there.

Hi there, is it possible to obtain the dask client when running with the `DaskExecutor`/`LocalDaskExecutor` or the

DaskTaskRunner

in Orion?

Morning - where there any issues with cloud this morning? I have a flow that failed (sent no slack message) - error was;

raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['secret_value'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

Hey team! With prefect 2 coming soon, would someone be able to point me in the direction of the benefits prefect cloud will have over hosting our own server? I notice the pricing page highlights the support and escalation, I was wondering if there was a similar breakdown to the technical benefits (for example, right now I know prefect cloud supports flow concurrency limits, while prefect server 1.x does not).

Hi all, how do we build docker image from orion branch ?

Executing task: docker build --pull --rm -f "Dockerfile" -t myprefect:latest "." <

Sending build context to Docker daemon  101.5MB
Step 1/38 : ARG PYTHON_VERSION=3.8
Step 2/38 : ARG BUILD_PYTHON_VERSION=3.8
Step 3/38 : ARG NODE_VERSION=14
Step 4/38 : FROM node:${NODE_VERSION}-bullseye-slim as ui-builder
14-bullseye-slim: Pulling from library/node
Digest: sha256:f7137af1e34927cb3251a6e091edff4592c49a89422196680a2f087edcbc6e4d
Status: Image is up to date for node:14-bullseye-slim
 ---> e2eb23c871f2
Step 5/38 : WORKDIR /opt/orion-ui
 ---> Using cache
 ---> 74138fe30a86
Step 6/38 : RUN apt-get update &&     apt-get install --no-install-recommends -y         chromium     && apt-get clean && rm -rf /var/lib/apt/lists/*
 ---> Using cache
 ---> 77731060b4e3
Step 7/38 : RUN npm install -g npm@8
 ---> Using cache
 ---> ac23c1b14bea
Step 8/38 : COPY ./orion-ui/package*.json .
When using COPY with more than one source file, the destination must be a directory and end with a /
The terminal process "/usr/bin/bash '-c', 'docker build --pull --rm -f "Dockerfile" -t myprefect:latest "."'" terminated with exit code: 1.

Hello all, I recently started experimenting with Prefect for one of my projects. I would like to ask whether it is possible to use the asynchronous functionality provided by dask; there are some tasks that involve IO output and as such they could be executed asynchronously (I am using the latest release of Prefect 2.0).

hey everyone, is there a way to use variables that are global to every task like an increment counter ?

Cross-posting here for visibility - this is a fantastic piece of content by Peyton that will: • walk you through getting started writing data pipelines in Python • showing challenges of operationalizing such Python script - the negative engineering part • walk you through how to turn this script onto a Prefect 2.0 flow • show you around the new Prefect 2.0 UI

Hey, was wondering if mapped tasks are able to output multiple values with nout ? i have an error when trying with dummy code:

from asyncio.log import logger
import prefect, pymongo, datetime
from prefect import task, Flow, flatten
from prefect.tasks.secrets import PrefectSecret
from prefect.tasks.mysql.mysql import MySQLFetch
from prefect.storage import GitHub

@task(nout= 3)
def produce_output(input):
    return input*2-1, input*2, ["val"+str(input*2-1), "val"+str(input*2)]

@task
def reduce(a):
    return sum(a)

@task
def retry_post(data):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(data)

with Flow("abcdef") as flow:
    success, fail, to_retry = produce_output.map([1, 2, 3, 4])
    total_success = reduce(success)
    total_fail = reduce(fail)
    retry_post.map(flatten(to_retry))

flow.run()

Hey, have a question: • Using Prefect 1.0 • Using

Local

storage and

Prefect Cloud

I want to use

stored_as_script=True

for my flow (pickle storage is not working with an SQLAlchemy object for some reason), and so I need to specify

path

— but that is giving me an error. Sharing the relevant code snippet and error.

with Flow(
        "eco_flow", 
        schedule=schedule, 
        storage=Local(stored_as_script=True, path='~/.prefect/flows/test-flow.py')
    ) as flow:

    cities = ['mumbai', 'new-delhi']
    
    for city in cities:
        data = get_traffic_data(city)  
        clean_data = clean_traffic_data(data)
        insert_to_db(clean_data)

Error after registering the flow:

Failed to load and execute flow run: ModuleNotFoundError("No module named '~/'")

I guess I am defining the

path

incorrectly?

Trying to use the DaskTaskRunner in Prefect 2.0 and getting this error. I am following the published documentation.

from prefect.task_runner import DaskTaskRunner
ModuleNotFoundError: No module named 'prefect.task_runner'

Is there an extra that needs installed?