I have a task that is returning an object with attribute

data

. But when I try to access this in the flow I get

'FunctionTask' object has no attribute 'data'

. How do I access the response.data?

with Flow('data_transformation') as flow:
     response = myTask()
     response.data //error

I have a question regarding Orion that is confusing me. With Prefect 1.0, we have the concept of flow storage. Is there an equivalent for Orion or how does it work? Say I want to run a flow on k8s, can I just package my flow code in a Docker container along with the required dependencies (maybe using the prefect image as base image)? Thanks for clarifying!

I'm new to prefect. Took some time to find out what Orion is. It seems to be a prefect 2.0? Should I, as a beginner go straight to Orion? Or stick to Prefect 1?

Also, I would be interested in seeing a comparison of Dagster and Prefect/Orion. Anyone aware of such a comparison.

Dumb question... how do I pass a parameter to

flow.run()

?

Hi all! We have an EKS cluster in which we would like to run our flows. We managed to build an architecture purely based on fargate but for our purposes we moved to an hybrid cluster, in which there are both managed nodegroups and a fargate profile. I would like to have a small node (identified by -ns prefect-agent) for my prefect agent which runs jobs on the fargate profile (identified by -ns serverless). I thought it was enough to specify the different namespace in the job_template of the KubernetesRun, by the way the flow run fails once started with a 400 error and the message: "the namespace of the provided object does not match the namespace sent on the request". I cannot find anything useful about running flows in a different namespace from the agent. Does anyone have managed something similar? Thanks in advance

For a KubernetesRun, I set a memory request, but see errors when the job needs more memory. i didnt set a memory_limit. is it possible to have it scale up automatically if I need more memory than the request?

Hi All, we have three environments: DEV,QA and Production and have corresponding docker images that contain dbt source codes. A prefect flow is used to run these dbt codes. The flow uses dockerrun. but dockerrun requires docker image name and label. and the image name and label are flow metadata. How can I use one flow for all these three environments?

Hello - my flows on prefect cloud keep failing. The state message for the failed run is

State Message: {'_schema': 'Invalid data type: None'}

but I am unclear on what that means. Could you clarify what this might indicate or ways to get more informative error messages? Thanks.

What are some ways that people execute flows remotely (not from the cloud UI)? If we wanted to execute a flow from a separate application what is the best way? Is there a webhook that we can call? CLI command to the agent?

Hi there! I have a question about task inputs and triggers.

Hi all! I keep running into an issue when registering a flow to prefect cloud. I can successfully do it when running via CLI but when I switch to registering through flow.register() i get

Invalid Project!

even though the project exists.

Hello folks, Is it possible to have a run with some tasks with status TriggerFailed, while there is no upstream Failed tasks? See screenshot below:

how do I make one task run after the other even though they do not have any edges between them?

Where and how do we store our applications settings in Orion?

Hi all! We have an issue with the prefect docker agent. We're using Azure and ACR with a system-managed identity assigned to vm. In order to login to acr we run every 3 hrs, there's a systemd timer for that:

az acr login

But, it seems that prefect docker agent only reads the token during start and stores it in memory. Unless we restart the prefect docker agent, it is unable to pull docker image flows after 3hrs (the acr token's expired)

Hello, I've been working with prefect the last months to see if it is able to replace some of our scripts I'm having an issue with the use of a prefect task built in. Postgres. The flow that I'm trying to achieve it's pretty simple, we just need to execute a query to gather some data ND then transform it and send it to another db. I'm not having a problem using the built in task but trying to connect with the rest of them as they don't appear in the UI as task... Thank you

Re:dbt features — Hi everyone, I look after Prefect’s partnerships and we're collaborating with dbt on what we should enable together. What would you like to see?

Hi Prefect community, I recently started to use Prefect and so far I mostly like it. Sometimes I can't wrap my head around how to apply the Prefect concepts. As in this problem I'm having now: I want to write a script that queries database tables and writes the data to a blob storage. The tables have to be queried with a number of different parameters (dates). So what I need is a product of tables x dates. I solved this by using a task named "create_input_tuples" that maps on the list of dates and gets the unmapped(tables) as an argument, too. Inside the create_input_tuples I loop over the tables and return a list of tuples (for each date) which I flatten then. Question 1: Is this the best approach or are there other ways to achieve that? After the queries have run, I filter for empty tables (which can happen) with a FilterTask and want to write the data from these tables to the blob. For this I need the table_name and date (for the naming). Question 2: Is there a way to keep track of the names and dates (also taking into account the filtering)? Usually I had a task returning a list of tables, to which I mapped the querying task, to which I mapped the load_blob task and the list of tables from the beginning. Without the filter this has always worked since the table_names and the tasks with the data from database aligned and could be mapped to the blob task. To make it a little clearer (I hope) I drew this picture, I wasn't sure if the Prefect visualization was really helping. Maybe you could help me find the prefect solution for this! Christopher

Hi there. Could someone help me understand why a flow (using Docker storage) is registered with a new version every time I run

prefect register

even though the code has not changed? The only thing I see that could change is the Docker tag which I am not setting explicitly. Would this cause it to increment the version number? Docker build log in thread.

Hi, I'm having the probably infamous issue with task slugs:

[2022-02-17 122919+0000] INFO - prefect.wait_for_flow_run | Flow 'affable-piculet-flow-build│

-dataset': Entered state <Failed>: Unexpected error while running flow: KeyError('Task slug do_someting-1 is not found in the current Flow. This is usually caused by a mismatch between th│

e flow version stored in the Prefect backend and the flow that was loaded from storage.\n- Did│

 you change the flow without re-registering it?\n- Did you register the flow without updating │

it in your storage location (if applicable)?')                                                │

when trying to run a flow of flows like this:

from flows.a import flow as flow_a   # flow's name is "flow-a"
from flows.b import flow as flow_b   # flow's name is "flow-b"

kwargs = {
"project_name": "foo",
"labels": ["test"]
}

flow_a.register(**kwargs)
flow_b.register(**kwargs)

with Flow("flow-of-flows", run_config=run_config, storage=storage, result=result) as flow:

   run_id_flow_a = create_flow_run(flow_name="flow-a", **kwargs)
   wait_for_flow_a = wait_for_flow_run(run_id_flow_a, raise_final_state=True)
   output_a = get_task_run_result(run_id_flow_a, task_slug="do_something-1")
   
   run_id_flow_b = create_flow_run(flow_name="flow-b", parameters={"input": output_a}, **kwargs)
   run_id_flow_b.set_upstream(wait_for_flow_a)

flow.register(**kwargs)

@George Coyne

Hi there. I am new to Prefect Orion. I am trying to schedule a flow run on my local Prefect server following the example POST request provided in the documentation. Here is my request:

POST: <http://localhost:4200/api/flow_runs/>

PAYLOAD:
{
  "name": "my_scheduled_flow",
  "deployment_id": "0ff88fc8-a71c-4b71-b8dd-078d6a36fb39",
  "flow_id": "94374bea-70ae-4e49-8988-e9077daa280a",
  "flow_version": "1.0",
  "state": {
    "type": "SCHEDULED",
    "name": "my_scheduled_flow",
    "message": "a message",
    "state_details": {
        "scheduled_time": "2022-02-18T16:10:49.005Z"
    }
  }
}

The flow run then appears in the server UI (as yellow) but never gets ran at the correct time. Does anyone have any idea on this? Am I missing an important part of the payload.

Anyone got this error today? Traceback (most recent call last): File "/usr/local/bin/dbt", line 5, in <module> from dbt.main import main File "/usr/local/lib/python3.9/site-packages/dbt/main.py", line 11, in <module> import dbt.version File "/usr/local/lib/python3.9/site-packages/dbt/version.py", line 11, in <module> import dbt.semver File "/usr/local/lib/python3.9/site-packages/dbt/semver.py", line 8, in <module> import dbt.utils File "/usr/local/lib/python3.9/site-packages/dbt/utils.py", line 9, in <module> import jinja2 File "/usr/local/lib/python3.9/site-packages/jinja2/__init__.py", line 12, in <module> from .environment import Environment File "/usr/local/lib/python3.9/site-packages/jinja2/environment.py", line 25, in <module> from .defaults import BLOCK_END_STRING File "/usr/local/lib/python3.9/site-packages/jinja2/defaults.py", line 3, in <module> from .filters import FILTERS as DEFAULT_FILTERS # noqa: F401 File "/usr/local/lib/python3.9/site-packages/jinja2/filters.py", line 13, in <module> from markupsafe import soft_unicode ImportError: cannot import name 'soft_unicode' from 'markupsafe' (/usr/local/lib/python3.9/site-packages/markupsafe/__init__.py) It seems that we have to downgrade markupsafe: Downgrading markupsafe to 2.0.1 fixes the issue on my side.

pip install markupsafe==2.0.1

Hello, I am new to Prefect. I want to structure my code so it can be run as Prefect workflow and optionally normal python program (Not as Prefect workflow). All functions has Task decorator and it can't be called like normal Python function. Please advise.

Hi is there any best practice for looping through tasks?

t1 = task1()
t2 = task2(t2)
while condition:
  t3 = task3(a)
  t4 = task4(t3)
  t5 = task5(t4)

Join us on February 24, 2022 at 11AM PST (2PM EST) to learn how to generate end-to-end lineage graphs of your data pipelines using Monte Carlo and Prefect so you can reduce the time to detection and resolution for critical data incidents: https://www.montecarlodata.com/how-to-build-more-reliable-data-pipelines-with-monte-carlo-and-prefect/

hey guys, i'm trying to get prefect running on kubernetes. i've packaged my entire python package into docker container, and i'm able to run the flows in the container just fine locally using

docker run

. however, when i deploy the flows to kube and use prefect cloud to trigger a run, i get the following error:

Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n  ModuleNotFoundError("No module named \'my_package\'")\nThis may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')

i'm using GCS as the storage. i'm also wondering if/why i need to use GCS as a storage -- doesn't the execution environment have access to the flows? why do they even need to be pickled? thanks in advance! 🙂

hello everyone! any chance that it is possible to use the DockerRun run_config with basic auth credentials?

Hi 👋, I am using

prefect.tasks.prefect.create_flow_run

to kick off some tasks. I've noticed that this isn't returning an id for the flow run as documented and instead returning a

Task

object. Any advice on how to access the ids of created runs?

Hi all! 👋 Going live for a fireside chat with @Jeremiah and @Chris White in 15 minutes. We'll be covering: 🕐 The Prefect 1.0 Release Candidate 🌠 Latest Prefect 2.0 (aka Orion) features including k8s support, dask support, filters and more (a ton has happened since our last Fireside Chat) ☁️ What's next for 2.0 You can join us on Youtube 📺, and feel free to ask any questions in the chat while you're there!