hello guys. I’m looking for a suggestion about Prefect 2.0 based on your experience. I was thinking which should be the ideal use case for workspaces. Initially I thought that workspaces could be useful to differentiate between enviroments, such as

development

and

production

, so to have a dedicated workspace for each environment. Then, after exploring Prefect 2.0 features, I’ve started to question this because one could use one workspace and setup everything in order to distinguish between development and production pipelines. So I wonder, what do you suggest for managing enviroments? Go with different workspaces or have one workspace? Then, what should be the ideal use case for a workspace?

Hi Prefect Community. I am currently trying to run a prefect flow using the DaskTaskRunner on a local kubernetes cluster. Unfortunately the flow is failing because of the following error:

File "/opt/conda/lib/python3.8/site-packages/distributed/worker.py", line 2742, in loads_function result = pickle.loads(bytes_object) File "/opt/conda/lib/python3.8/site-packages/distributed/protocol/pickle.py", line 73, in loads return pickle.loads(x) ModuleNotFoundError: No module named 'prefect'

When entering the container, prefect seems to be available though. Do you have any idea what could be the reason?

I wanted to double-check before drafting up the wrong solution, but it looks like a flow of flows doesn't need to be in the same project as long as create_flow_run can find the flow_name within a project? https://www.prefect.io/guide/blog/flow-of-flows-orchestrating-elt-with-prefect-and-dbt/. We're considering organizing all of our API calls in a specific project and DBT flows within a DBT Project.

Hi Prefect community, I am facing issue while registering flows in prefect and getting below error, Can anyone please help me with this.:

botocore.exceptions.ClientError: An error occurred (AccessDeniedException) when calling the GetSecretValue operation: User: arn:aws:sts::xxx:assumed-role/code_deployments-role/iddoc- is not authorized to perform: secretsmanager:GetSecretValue on resource: 4i-adl-config because no identity-based policy allows the secretsmanager:GetSecretValue action

👋 Is anyone scheduling DBT with Prefect here? How do you deal with dependencies (upstream & downstream)? Do you handle the logic of "This model needs to run before this other model"? On our side, we want to leverage the DBT built-in features as well as the visualization of Prefect with 1 DBT model == 1 Prefect flow

Hi all! Come join @Khuyen Tran, @Chris White and myself on PrefectLive starting now!

Question on Prefect 2.0... how is work on a

map

operator coming? That is a big stumbling block for us adopting 2.0. Thanks!

Hi y’all. I’m trying to sign up for a 2.0 cloud account with my work email (I already have a cloud 1.0 account). I’m getting the error

AxiosError: Request failed with status code 401

. So I thought maybe I could sign in with my 1.0 credentials and got

Error: Invalid username or password.

Am I unable to sign up for 2.0 if I also have 1.0?

Hello. I'm using

subprocess

to run my dbt project locally (Prefect 1.2.2, Windows). I create my command (

dbt run --vars '{data_processing_start_date: 2022-07-20, data_processing_end_date: 2022-07-20}' --profiles-dir ./

) like this:

command = (
        f"dbt run --vars '{{"
        f"data_processing_start_date: {data_processing_start_date}, "
        f"data_processing_end_date: {data_processing_end_date}}}' --profiles-dir ./ "
    )

The command is created correctly, but dbt gives me this error

dbt: error: unrecognized arguments: 2022-07-20, data_processing_end_date: 2022-07-20}'

Seems like dbt interprets

2022-07-20

as an argument instead of the value for

data_processing_start_date

variable. Keep in mind that when I run the same command in my CLI, dbt works fine. But when it's provided to dbt through

subprocess

this occurs. This is my subprocess:

subprocess.run(
        command,
        check=True,
        stderr=True,
        stdout=True,
        shell=True,
        cwd="dbt",
    )

Any idea what might cause this, and how to solve this? Thank you

Hello! Could you please check if this issue was fixed - {https://github.com/PrefectHQ/prefect/issues/5663} Thanks!

Hi there! Sometimes our tasks get hung (attaching screenshots for the same), they remain in pending state when their upstream tasks are completed hours before and even after cancelling/restarting em, they won't start remain in pending state. This is happening with us like 3rd time, any possible reasons why it happens? or it is a bug in prefect 1.0?

Did something change in 1.0? My Agent died, and this is how I always started it.

Hi Community! Is there a way to deploy the whole repo and not just 1 flow file with GitHub storage?

Hi. How to pass job template in KubernetesRun as python dict?

{
            'apiVersion': 'batch/v1',
            'kind': 'Job',
            'spec': {
                'template': {
                    'spec': {
                        'nodeSelector': {
                            "<http://topology.kubernetes.io/zone|topology.kubernetes.io/zone>": "us-east-1a",
                            "<http://dask.corp.com/subnet-type|dask.corp.com/subnet-type>": "private",
                            "<http://dask.corp.com/storage|dask.corp.com/storage>": 'true',
                        },
                        'containers': [
                            {'name': 'flow'},
                        ]
                    }
                }
            }
        }

Hi prefect team, Is there a way in prefect to set a "universal retry" policy for all flows in prefect 2.0? For instance if you didn't specify a retry option in your code it would automatically take this universal setting. If a retry policy is set in code then that would override the universal setting. Second question is there a way to set unlimited retries, besides just setting a really big number for "retries" parameter?

Hello Team, I am using prefect 1.x. What is the best practice for prefect flow code deployment with git repo version control instead of keep registering flow on code changes.

Hi folks, 1. Is there any way in prefect 2.0 I can say do

task1

. But do

task2

only if

task1

is completed, and

task3

only if

task1

and

task2

are complete? 2. Does prefect 2.0 parallelize flows over

async for

s? 3. I write (append) to a file in one of my tasks. Is that

prefect

safe?

hey prefect team, i'm excited about the new 2.0 releases and was wondering if there is already a release note for 2.0b10 available. which version is running in prefect cloud and are there any breaking changes? unfortunately the

admin/version

api route doesn't work for me. thanks a lot.

Hi Community, using Prefect to schedule dbt jobs, is it necessary to have run_id as a variable in dbt run command?

Hi, community ✌️ After upgrading to prefect==2.0b10, I get the following error:

KeyError: "No class found for dispatch key 'S3 Storage:sha256:68ed [...]' in registry for type 'Block'."

. The full error trace and code can be found in the first comment to this post.

so flows work locally but

__file__

is not defined when prefect cloud runs the flow. Any idea what the issue is?

👋 Hi, I’m new here, trialing Prefect 2.0, and have a couple of question. I’m writing a flow to copy data into a bunch of Snowflake tables. I wanted to use the Snowflake connection instead of managing a connection myself. Seems like

snowflake_query

and

snowflake_multiquery

are their own tasks, so I can’t use them as part of a different task? Which means that my flow looks like this:

do_some_stuff()
for table_name in tables:
    queries = build_queries(table)
    snowflake_multiquery(queries)

These tasks are all running sequentially, but there are a lot of tables so I’d like the

snowflake_multiquery

tasks for each table to run concurrently. How can I make that happen?

Second question:

snowflake_multiquery

was throwing

Object of type SecretStr is not JSON serializable

errors from this line of code in the connector, where it dumps the request body, including connection auth info, into the request. I got around this for now by changing the password here from type

SecretStr

to

str

, but that’s obviously not a great solution 😅 Has anyone else run into this issue?

Hi, I am in the process of evaluating Prefect Cloud for my pipeline. I have a setup where there is a flow definition with parameters, and I have about 1000 parameter sets from database (and a lot more when the pipeline is live). What I can do is to deploy the flows with parameters for each parameter set via API. I want to know how this scenario can be handled: If I update the code of my flow, such as using a different algorithm, or add another step, how can 1000+ deployments be updated?

👋 i see a new prefect version (

2.0b10

) was released yesterday, but don't see associated release notes. was this just a minor bug fix?

When using prefect 2.0, how do I verify that agents are able to communicate with the cloud? On 1.0 there was an agents screen thats showed all connected agents.

On Prefect 1.0, does the agent run each flow in its own process? I think the answer is yes

👋 another qq: is there any kind of documentation on the type of data the prefect agent might send back if we were to use prefect cloud vs self-hosted?

On 1.0, is there a way to run a ShellTask where the stdout/stderr is streamed in its raw format, bypassing the prefect logger?

Hey all! can i make an

@task

that across all concurrent flows is a Singleton (e.g only one of that task can run at a time)?