Hi! Why does this example lead to the creation of persisted results in the results folder? Thanks 🙂

@task(checkpoint=False)
def extract_things():

    bar = [1, 2, 3, 4, 5, 6, 7, 8, 9]
    foo = ['a', 'b', 'c']

    data = {
        'bar': bar,
        'foo': foo
    }

    return data


@task(checkpoint=False)
def do_things(data):

    for x in data:
        print(x)

    return None

with Flow(
    "Do it",
) as flow:
    things = extract_things()
    do_things(things['foo'])
    do_things(things['bar'])

Hello from Chicago! 👋 Could someone confirm the right mental model for uniqueness in Prefect Cloud? • `project`: name is unique across all projects within a tenant, id is globally unique across all of prefect cloud • `flow`: name is unique across all flows within a project, id is globally unique across all of prefect cloud • `tenant`: id is globally unique across all of prefect cloud I'm writing stuff that interacts with Prefect Cloud, and I want to be sure that my assumptions here are correct. So, for example, I want to be sure that two flows called

c00l-flow

could exist in the same tenant, as long as they are in different projects. thanks!

Is there a way to specify that the kubernetes agent should set certain environment variables (particularly from a kube secret) in the pods that it creates?

Following up on subflows. I was wondering if this is still an active area of development---I know I asked about this before (https://prefect-community.slack.com/archives/CL09KU1K7/p1586212582026500), but it looks like there hasn’t been activity on the linked github issue in a little while. It would be very valuable for us to have subflows as first class objects. One major use case is that we’d like to have a single production version of a flow, and then we’d like for other code to be able to modify that flow (e.g. by swapping out a large subgraph for another subgraph that has the same keyed upstream and downstream edges). It seems to me that the most natural way to achieve this is to have nested subgraphs of tasks within a flow. I’ve toyed around with an implementation that I think would work, and would love to know if there is any appetite from all y’all at

prefect

for something like this. The idea is that currently

Flow

objects have two major responsibilities: (1) building/maintaining a DAG and (2) supporting execution of that DAG. I propose splitting that functionality into two classes:

Flow

, which would support DAG execution, and

Composite

(or some other name), which builds and maintains a DAG. Crucially,

Composite

instances could be nested (i.e.

Composite.add_task

would accept either a

Task

or another

Composite

instance, and similarly for

replace

,

add_edge

, etc). To run the DAG,

Flow

would traverse the nested graph and build up a flattened DAG that just contains

Task

nodes, and then execute as usual.

Hi Everyone, new to prefect, looking for some pointers/guidance how to run PowerShell with prefect. I started looking at using the shell task but, I am not having much luck. My agent is running a Windows Server 2019. Thanks in advance.

hey quick question: I want to be able to send data to my database over the course of a flow run in order to update multiple different database entries for objects that are created/used during various tasks within this flow. What I'm planning to do is create a database object at the beginning of my flow and store it in my flows context. I'll probably put this functionality into a task. Is there a better way to do this? This way throughout the flow I'm able to send/update database entries by grabbing a reference to this database connection from the context, or maybe just using it directly. What I really want to be able to do is have this database be a property of the flow, so that I can have a Flow-database-object referencable across all my tasks, but that doesn't seem to be possible at the moment, and the context variable seems to be the best way to do this. Another solution I just thought of is to break all these database connections into smaller tasks, and have each task have a callback function but then I'll essentially have WAY to many tasks in my flow. I'm curious if others have had this problem, and what solutions they had.

I user long-descriptive names for my flow runs, any chance the UI could get an update to accommodate that?

Prefect Cloud and the Prefect Cloud API are both affected by what appears to be a 3rd party outage at the moment. We will keep you updated as we can

this issue appears to be resolved, or partially resolved. We will continue to monitor and update you as we can

this issue may or may not be tangentially or directly related to our outage https://www.cloudflarestatus.com/

hey - what's the best way to diagnose why an agent might not accept a flow? I have a scheduled run of a flow tagged with my local machine name, and I'm running an agent with the label. Is there any debug logging for the agent to show what flows it's considering? (I should mention I don't think this is related to the agents, as the UI says it's querying)

Hi folks, is the project feature only available in Prefect Could? I cannot create project in Prefect core.

Hi folks, I encounter another issue: I’ve created a flow template and register multiple flow instances with different parameters. But I cannot find a way to retrieve back those parameter details from those flows. Anyone can help on it?

Question about setting prefect environment variables if I set the environment variable prior to running the script then it works fine

export PREFECT__LOGGING__LEVEL="ERROR"

but if I set the environment variable from within the script - prefect doesn’t pick up on it

import os

os.environ["PREFECT__LOGGING__LEVEL"] = "ERROR"

....

flow.run()

what am I missing here ? (is there a way to set the variable programmatically from within the script? )

From the docs on results https://docs.prefect.io/core/advanced_tutorials/using-results.html

A result subclass that matches the storage backend of your prefect.config.flows.storage setting will automatically be applied to all tasks, if available; notably this is not yet supported for Docker Storage

Does it mean I have to individually configure the results of tasks in a docker stored flow (without any global defaults) or does it mean it is currently unsupported to have any results for tasks that run in a docker environment?

Hi. I have a distributed Dask setup (DaskExecutor). I learned that I need to persist flows (and results) to be able to survice prefect cluster restarts. I added an NFS volume to share the exact same data between Agent and Dask Worker (these two only!) and added a

Local

storage to the flow, pointing to a path in the NFS. Now the questions: 1. is it a good idea (or necessary?) to have the flows and results accessible in the same mountpoint (local filesystem location, e.g. NFS-persisted

~/.prefect/flows

) for Agent and Dask worker? Or is it sufficient to persist the flow/result storage (make it durable between restarts), and its not needed for the agent to access the pickeld flow results? 2. What's the pro and con of

stored_as_script=True

for a

Local

storage? When do I want to set this? Also, I do not understand the

path

parameter for local storage? I set

directory

to a path on my NFS and now I see the pickled flows and results. What is path?

We are using a FargateAgent and currently experimenting with different memory and CPU settings for the flows it fires off. We have the issue that we can’t change the resources anymore because it fails to recognize that it needs to recreate and ECS TaskDefinition when the memory was changed. So even with setting the right values it starts the old task definition with old memory settings:

class ECSFargateAgent:
    def __init__(self, config: PrefectAgentConfig):
        self._env = config.env
        self._agent = FargateAgent(
            labels=[f"reco-{self._env}"],
            enable_task_revisions=True,
            launch_type="FARGATE",
            taskRoleArn=config.task_role_arn,
            executionRoleArn=config.execution_role_arn,
            cluster=config.ecs_cluster_arn,
            networkConfiguration={
                "awsvpcConfiguration": {
                    "assignPublicIp": "ENABLED",
                    "subnets": config.subnets,
                    "securityGroups": [config.security_group],
                }
            },
            cpu="1024",  # 1 vCPU
            memory="3072",  # 3 GB
            containerDefinitions=[
                {
                    "logConfiguration": {
                        "logDriver": "awslogs",
                        "options": {
                            "awslogs-group": config.log_group,
                            "awslogs-region": "eu-west-1",
                            "awslogs-stream-prefix": "flows",
                        },
                    },
                }
            ],
        )

    def run(self):
        """Start the agent"""
        self._agent.start()

e.g. we switched

memory

from 16GB to 3GB but the agent still fires off flows with 16GB. Only deleting the old task definitions manually solves this for us. Anybody seen something similiar or has a solution?

# prefect run server -n flow2 -l
Flow Run: <http://localhost:8080/flow-run/e82df5c1-d197-46f1-9bb1-f18f9335e0f8>
TIMESTAMP                         LEVEL    MESSAGE
2020-07-20T12:09:56.284609+00:00  INFO     Submitted for execution: PID: 835
2020-07-20T12:09:57.237871+00:00  INFO   Beginning Flow run for 'flow2'
2020-07-20T12:09:57.273447+00:00  INFO   Starting flow run.
2020-07-20T12:09:57.273836+00:00  DEBUG  Flow 'flow2': Handling state change from Scheduled to Running
2020-07-20T12:09:57.581024+00:00  INFO   Task 'task2': Starting task run...
2020-07-20T12:09:57.724552+00:00  INFO   hello, task2
2020-07-20T12:09:57.835204+00:00  INFO   Task 'task2': finished task run for task with final state: 'Success'
2020-07-20T12:09:57.942237+00:00  INFO   Task 'flow1-runner': Starting task run...
2020-07-20T12:09:58.145036+00:00  ERROR  Unexpected error: HTTPError('400 Client Error: Bad Request for url: <http://localhost:4200/graphql>')
                                         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/task_runner.py", line 819, in get_task_run_state
                                             value = timeout_handler(
                                           File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 188, in timeout_handler
                                             return fn(*args, **kwargs)
                                           File "/usr/local/lib/python3.8/site-packages/prefect/utilities/tasks.py", line 445, in method
                                             return run_method(self, *args, **kwargs)
                                           File "/usr/local/lib/python3.8/site-packages/prefect/tasks/prefect/flow_run.py", line 119, in run
                                             flow_run_id = client.create_flow_run(
                                           File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 877, in create_flow_run
                                             res = self.graphql(create_mutation, variables=dict(input=inputs))
                                           File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 213, in graphql
                                             result = <http://self.post|self.post>(
                                           File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 172, in post
                                             response = self._request(
                                           File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 328, in _request
                                             response = self._send_request(
                                           File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 265, in _send_request
                                             response.raise_for_status()
                                           File "/usr/local/lib/python3.8/site-packages/requests/models.py", line 941, in raise_for_status
                                             raise HTTPError(http_error_msg, response=self)
                                         requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <http://localhost:4200/graphql>
2020-07-20T12:09:58.275483+00:00  INFO   Task 'flow1-runner': finished task run for task with final state: 'Failed'
2020-07-20T12:09:58.302057+00:00  INFO   Flow run FAILED: some reference tasks failed.
2020-07-20T12:09:58.302667+00:00  DEBUG  Flow 'flow2': Handling state change from Running to Failed

During this flow run the log/stdout of Apollo (wich is listening to

localhost:4200

) prints

I cannot start a flow run from another flow. Can you please verify my minimal example? I have two flows. This is registered as

flow1

:

#!/usr/bin/env python
# coding: utf-8

import prefect
from prefect import task, Flow
from prefect.environments.storage.local import Local


@task
def task1():
    prefect.context.get("logger").info("hello, task1")


with Flow("flow1", storage=Local(directory="/flows/.prefect/flows")) as flow:
    task1()

if __name__ == "__main__":
    flow.register()

And here is

flow2

:

#!/usr/bin/env python
# coding: utf-8

import prefect
from prefect import task, Flow
from prefect.tasks.prefect import FlowRunTask
from prefect.environments.storage.local import Local


@task
def task2():
    prefect.context.get("logger").info("hello, task2")


with Flow("flow2", storage=Local(directory="/flows/.prefect/flows")) as flow:
    flow1_run = FlowRunTask(name="flow1-runner", flow_name="flow1", wait=True)
    task2.set_downstream(flow1_run)

if __name__ == "__main__":
    flow.register()

When I now run the

flow2

I get an error:

2020-07-20T12:09:58.139Z {"message":"Variable \"$input\" got invalid value { flow_id: \"80de1b39-fed7-47ab-a5a5-fc22ac6f87d2\", idempotency_key: \"e82df5c1-d197-46f1-9bb1-f18f9335e0f8\" }; Field \"idempotency_key\" is not defined by type create_flow_run_input.","locations":[{"line":1,"column":10}],"extensions":{"code":"INTERNAL_SERVER_ERROR"}}

Is this a bug? Every component of my Prefect Cluster (v0.12.4) is running in K8S but its a static setup with every component is one container and everything together in one pod (so everything can communicate using localhost). I have a Dask-Worker pod running and all other flows (not calling other flows) this far are working.

Hey guys, I am trying to create_flow_runs programatically using

from prefect.client import Client
import pandas as pd

if __name__ == "__main__":
    dates = pd.bdate_range(pd.Timestamp("20200601"), pd.Timestamp("20200630"))

    c = Client()
    for d in reversed(dates):
        c.create_flow_run(
            "e04181f3-86b3-4f6f-87cf-ed7c24cb95ec",
            parameters=dict(date=d.strftime("%Y-%m-%d")),
            context=dict(name="John Doe"),
            run_name=f"test_backfill_{d.strftime('%Y%m%d')}")

But some of the flow runs are stuck with

Queued due to concurrency limits. The local process will attempt to run the flow for the next 10 minutes, after which time it will be made available to other agents.

I am on the dev version of the cloud backend and understand that there are limits. I have the following questions: 1. Am I hitting this error because of the dev cloud version? 2. Although, the message says the flow will be rescheduled within the next 10 minutes, it takes around 30 mins for the rest of the tasks to get scheduled and run. 3. Is there a better way to create a backfill? I am avoiding

FlowRunTask

because context is currently not passed to the triggered flow runs

Hello guys, this should be fairly easy but i dont know the most "Prefectish" way of doing it. Imagine we have a Task do_something_big that is too big and we want to separate it into a few different methods. Is it possible to have a main Task that calls other tasks inside it (that are apt for dask's parallelism)? My first approach of having them all in the flow definition as flow_part_1, flow_part_2, flow_part_3 (depending on eachother) works but I'd rather have a much simpler gantt chart being generated out of it with 2 or 3 (at most) main Tasks. As an example:

@task
def i_am_too_big():
"""
multiple
 lines of code
 that should
 go into 
a different method
"""

"""
some
sequential (to be parallelized)
http
calls
""""

"""
multiple
 lines of code
 that should
 go into 
a different method
"""

Hello guys, I'm trying to move to Prefect from airflow and was wondering if there was anything like Airflow's subdagoperator in Prefect?

Hello everyone. Is there an API function to get the id of the latest version of the flow, given the project_name and the flow_name?

Hi guys, I hit the error in agent:

ERROR - agent | Error while deploying flow: ValidationError({'_schema': 'Invalid data type: None'})

What I did is creating the workflow with a label and using graphql API to register the flow:

alert_flow.environment = LocalEnvironment(labels=['public'])

create_flow_mutation = """
            mutation($input: create_flow_input!) {
                create_flow(input: $input) {
                    id
                }
            }
        """
        serialized_flow = flow._flow.serialize(build=False)
        self._client.graphql(query=create_flow_mutation,
                             variables=dict(
                                 input=dict(
                                    serialized_flow=serialized_flow,
                                    set_schedule_active=False
                                    )
                                )
                            )

I do not get more log for debugging, please help, thanks!

Hi Guys, was curios about what would happen if I have a single server agent stack and had hundreds of scheduled flows that run at the same time. How would concurrency be achieved in a docker based architecture? Will it depend on the number of cores the docker host has or do we need to do something else to achieve concurrency?

Hey all - I am a bit confused by this error, raised randomly by a scheduled flow:

Failed to retrieve task state with error: ClientError([{'path': ['get_or_create_task_run'], 'message': 'Invalid ID', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])

Is this raised in GraphQL? What are the possible reasons? I am using Cloud + Core 0.12.1

Hi all. Is there a way to get more descriptive tracebacks i Prefect Cloud? Currently I'm getting:

Unexpected error: ZeroDivisionError('division by zero')

Traceback (most recent call last):

File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner

new_state = method(self, state, *args, **kwargs)

File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 952, in get_task_run_state

self.task.run, timeout=self.task.timeout, **raw_inputs

File "/usr/local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 182, in timeout_handler

return fn(*args, **kwargs)

File "<ipython-input-6-39643be0b44f>", line 9, in hello_task

ZeroDivisionError: division by zero

I know it is part of the Hybrid model that code isn't shared but is there a way to get a normal python traceback, so I can see the failed line?

https://docs.prefect.io/api/latest/environments/storage.html#local <- is

stored_as_script

and

path

keyword arguments removed? in every storage class? (

Local

in my case)