Hey All, I'm setting up a new instance of Prefect and I seem to be misunderstanding something about flows. I've got a bunch of methods with @task decorators and then I set them up in what should be a single a flow. When I load this into Prefect though I get two split DAGs that seem to function independently. Flow setup here and picture of the result attached.

with Flow(flow_name) as flow:
    <http://logger.info|logger.info>(f"{flow_name} Task Flow initiated, running in {file_path}")
    df = pull_oracle_data_via(oracle_query_sql=oracle_query_sql_path,prod=use_oracle_prod)
    df = set_data_types(df)
    upload_to_table(df, destination_table = data_destination_table_name)

    if summary_view_name is not None and history_table_name is not None:      
        <http://logger.info|logger.info>("Initiating history upload process.")
        summary_df,summary_data_empty = pull_summary_data_via(sql=f"SELECT * FROM {summary_view_name}")
        if summary_data_empty:
            delete_today_from_history_if_exists(df=df,history_table=history_table_name)
            upload_to_history_table(df=summary_df, destination_table=history_table_name, append=True)
    else:
        <http://logger.info|logger.info>("Skipping summary view run: summary view name and/or history table name missing.")

To address this I tried to make the dependencies explicit, by adding "upstream_task" flags to two of the above lines as so:

summary_df,summary_data_empty = pull_summary_data_via(_upstream_tasks_=[upload_to_table],_sql_=_f_"SELECT * FROM {summary_view_name}")

delete_today_from_history_if_exists(_upstream_tasks_=[pull_summary_data_via],_df_=df,_history_table_=history_table_name)

This doesn't seem to fix the issue though; when I run the flow, later tasks still seem to initiate before the Oracle pull, which should occur before everything. Anyone see what I'm doing wrong? The documentation would seem to indicate that feeding result data from one task to another would make dependencies work correctly, but that doesn't seem to be happening here.

hi, maybe I’m tired today, but I can’t figure out if I can set / change the context variable inside of the Flow block.

Hi, anyone has tried to use shelltask to run powershell scripts?

Does prefect have a mobile app?

Is there a good way to read in custom SQL, then delete 14 days of data from the target table, then deposit that custom SQL for the last 22 days into the target table, ensuring no duplicate records? Something like

extract => transform => delete 14 days in target => append last 22 days w/ no dupes

. Right now I"m doing 2 parallel ETLs, one which ends in replacing the existing data and the other which appends. This seems very slow since I have to perform 2 full extracts.

I'm doing some concurrency testing using a simple workflow with a mapped task that starts (dummy) AWS Batch jobs. Running locally (flow.run()) I manage to run into the AWS API limits. When I run the same flow via Prefect Cloud and an agent the bottleneck seems to be somewhere else because I don't see more than about 100 tasks running concurrently. In both cases I use a LocalDaskExecutor with 500 threads and the EC2 instance running the flow/agent is not at all busy (only waiting for Batch jobs to finish after all). Any ideas what could cause the slowdown using Cloud?

Hi, I pip installed orion and tried to start the server:

prefect orion start

- some error messages appear, including sqlite

no such table 'flow_run'

- is there some db initialisation procedure that I missed?

How can I open the 4200 port to all interfaces? It runs on a server an not on my laptop

I'm having several issues setting up a logger for my shared code; I've tried to follow the docs. 1. I've set the ENV

PREFECT__LOGGING__LEVEL="DEBUG"

within the run_config (ECSRunner) and I've confirmed that this env is set as such in the running container. However, within my running flow, when I check

os.environ

its set to INFO 2. I setup a logger in my shared code that I later import like a library:

logger = logging.getLogger("dwh.utils.main")

. I've set

PREFECT__CLOUD__SEND_FLOW_RUN_LOGS='true'

and

PREFECT__LOGGING__EXTRA_LOGGERS="['dwh']"

. when I execute a flow that uses that shared code, I can't see it emit logs in CloudWatch or in Prefect cloud, however when I run the flow locally I do see the logging statement what am I missing?

Hello! An Orion question here. At the moment, I’m uploading wrangled Pandas DataFrames as a parquet file to S3 as it’s own task. I would like to refactor away these tasks and use S3Result with PandasSerializer instead. My question: will the Results API in its current v0.15 form (ie as separate classes to be instantiated, given an appropriate serialiser and passed into the task decorator as a keyword argument) follow through into Orion?

Hi everyone, first post here and just coming up to speed on Prefect. So far it's a joy to use! My question is about running an agent on AWS, hopefully something like ECS. Specifically I'd like to have a durable setup/flow that stays on AWS, and totally cut my local computer out of the loop for everything. I see a lot of good tutorials that involve initiating a flow with a local computer that runs on AWS using ECSRun. (like https://towardsdatascience.com/serverless-data-pipelines-made-easy-with-prefect-and-aws-ecs-fargate-7e25bacb450c and https://www.lejimmy.com/distributed-data-pipelines-with-aws-ecs-fargate-and-prefect-cloud/) Tantalizingly, the second post by Jimmy Le has one mention that sounds close to my use case: "Ultimately we'll probably want to run many ECSAgents in a small Elastic Compute Cloud (EC2) with

supervisord

monitoring their health." What I'd like and what my question is: Is there documentation on just setting up a Docker image and having that be self-contained container running a flow on AWS, without needing to kick it off from a local computer? I'd be happy to use Prefect Cloud for UI. Is this a valid use case? Is there some reason there's not much documentation for it? (Or do I just not understand something basic?) thanks so much for any help.

Hello, I want to use SnowflakeQuery Task against Snowflake configure with Okta SSO and MFA enable. Based on the documentation on Snowflake connector documentation I would have to pass

authenticator

to the connect method (which doesn't seams to be possible at the moment in the task). Is my assumption correct or is there a magic env variable I can use? If not is it a change I can request or implement my self in the repository?

Hello, please I am having some issues getting my prefect flow to run. I am trying to create a parent flow to run other flows based on a conditional value that will be derived from the file dropped in s3 and the column it matches in a snowflake table. I created a task that extracts the column value we are looking for from the snowflake table, then used if statements to assign the flow to run based on the value extracted from the table. I then return the flow to run, so I can pass it as the flow name in the startflowrun function. it seems I'm unable to pass the flow to run as it always returns not defined. also when I tried to just test out the startflowrun with an existing flow, it returns an error that the flow is not found. Please is there a way to get this job name and pass it as the job name for startflowrun to run?

Can prefect be used to extract from a database and do an SFTP upload? Are there examples of this?

Hi everyone, we just updated from 0.15.4 to 0.15.9 and suddenly our flows are failing on a specific line of code. Every instance of

if status in ['SUCCEEDED', 'FAILED']:

fails with the following error:

TypeError: 'sequence' not supported between instances of 'str' and 'tuple'

If we log the type of string and array, it returns their types correctly.

Hello ! 👋 Just a quick question, I have a task

B

which depends on a undefined size list of other tasks

[A1, A2, A3, ...]

(Prefect therefore creates an implicit

List

task under the hood). The thing is, some tasks in the list

[A1, A2, A3, ...]

can be skip at runtime, but I still want

B

to be executed. I currently can’t achieve this, even if

skip_on_upstream_skip=False

is specified for

B

, since the implicit

List

task is skip without being able to do anything (I receive

None

, and not a list of optional elements). Is there a way to do it? Thanks!

Hi there, I am running into a weird issue where if I provide a

dockerfile

to the

prefect.environments.storage.Docker

class, I get an error that looks like this:

shutil.Error "[Errno 63] File name too long: ['<Endless recursive path>']

Hey Folks, question about SLAs - is it possible to trigger a flow run in case a certain flow run fails ?

When creating re-usable flows, how do you name them uniquely? For example, if I have a flow that connects to an SFTP, Downloads those files, and then writes them to cloud storage based off passed parameters - how does the flow name align with the parameters being passed?

Hi I encountered interesting issue when running the same flow using local agent vs a k8s agent, it seems like the same code takes longer to run on local agent compared to the k8s job [all on the same cluster, machine type, and resource requirements]. is there anything on local agent configuration or run config that can explain the difference? it runs x4 slower

I've managed to silence the logger: I can't see the stacktrace when a task crashes. How do I make Prefect print the stacktrace locally?

@task(
    nout=2,
    max_retries=10,
    retry_delay=timedelta(seconds=1),
)
def fetch_model_settings(
    dsn_params: DSNParams,
    app_id: UUID,
    default_model_group_id: UUID,
) -> ModelSettings:
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(
        f"Fetching model settings for app_id={app_id}, default_model_group_id={default_model_group_id}"
    )
    raise ValueError("WTF where are the logs")

# before flow:
prefect.config.logging.level = "DEBUG"

In the sample above I never get to see the WTF in the output of running the flow in the console/locally (

PREFECT__LOGGING__LEVEL=DEBUG python flows/flow.py

)

In one of my flows, I have a task to upload a video to s3 (boto3 s3 resource

upload_file()

call, because big&binary). Since it's synchronous, I can trust the file will be where desired when task succeeds. Next task in this flow is to transcode this video. I do so by creating a

mediaconvert

boto3

client, and sending a mess of json to its

create_job(**args)

method. This returns me... the job. Now from what I've read... I should be able to use a prefect built-in

prefect.tasks.aws.client_waiter.AWSClientWait

to wait for said job to finish (which is fine, at this point the workflow is serial/synchronous). Problem is... even when the job reports success (in the console, even!), it takes a while (minutes?!) for the transcoded movie to be present in the target bucket. I would then... need to enter another wait task until I could find the file in the bucket's list of objects, possibly through prefect.tasks.aws.s3.S3List? until I could proceed to do further things to this transcoded video? This conjunction sounds all too common not to have an integrated solution, unless I'm being dense (hah! no news there!) and not spotting an obvious solution. Any guidance?

Hello! Q about auth in Cloud. If I register by invitation via Google, can I set new password and login by email/pass?

Hello! I'm considering to use Prefect to orchestrate VM as I work on a Cloud called WEkEO Dias and does not offer dynamic provisionning of instances. This typical workflow would be executed on a master VM : 1. start a VM 2. launch a flow run on this VM 3. shutdown the VM I thought about using the function

create_flow_run

to launch part 2 synchronized with steps 1 and 3. Considering I can configure some script to deploy the Agent on VM start, is it possible ? I'm concerned about the main flow run starting before the VM's agent exists If you have suggestions on better ways to implement this with Prefect I'm interested ! So far i only used basic ECS agent 🙂

Hey everyone, I had something scheduled via the Prefect CronClock and Scheduled (prefect v0.15.5), and noticed that turning off the schedule on the cloud UI, it still ran at the scheduled time. That's not what I expected to happen, has anyone else experienced this?

Hello again! I'm back to helping our science guys use prefect. I went to team->members to invite someone, and it says we're out of accounts, and offers a link to https://cloud.prefect.io/team/account which is broken. I got to the page, though, and apparently we need an enterprise account for more users. I'll submit the form, but it would be great if we could get unblocked today

Hi there! I am new to prefect and looking whether it is the right tool for our needs in our Data Engineering team. Therefore, I wanted to first run the simple tutorial described here: https://cloud.prefect.io/tutorial/Universal-Deploy#user-content-universal-deploy I am working on Windows on WSL1. Everything worked fine (creating python environment with prefect, running the test flow locally, authentication with the key, creating the project), no error message, but I can not see the flow I defined in the UI within prefect cloud. What could be the reason for it?

Hey Everyone - For those using DbtShellTask, is it expected to have to run

dbt compile

before executing

dbt run

?

Our data engineering team is considering the following setup: • snowflake as the database • s3 buckets as storage • prefect or airflow as orchestration tool For airflow AWS offers MWAA which basically creates the whole infrastructure (scheduler, UI through webserver, worker)needed for orchestration. For prefect I would think that we could use prefect cloud, which handles the ui/webserver aspect, but we would still need to setup a server running an agent, right? As we would like to not manage any server in order to keep the infrastructure as simple as possible: What would be the best option? Or did I misunderstand something? Thanks for advice in advance!