Hey P team, a question about dask interaction. We are looking for best practices for interrupting and then restarting a flow. Our use case is that we need to sever all database connections during a database migration. We would like to avoid restarting the flow from the beginning, but we are fine with re-starting any running tasks from the beginning (in fact I don’t think we have a choice there). We are thinking of the following steps: • pause all running tasks using graphql API • scale down dask workers to 0 (we run in fargate) • scale up dask workers after the migration • unpause the tasks Would love to hear any thoughts. Thanks! 🙏

We’ve got about a third of the Prefect team and @Nate Atkins chatting over in the Core Cantina right now if anyone wants to join — @Jeff Brainerd that’s a great question, let’s discuss if you can make it! https://prefect-community.slack.com/archives/C0106HZ1CMS/p1597433468091100

I have a quick question: how can I set a flow_run_id within the call of a

FLOW_OBJ.run()

? I'm currently doing the following:

with prefect.context(flow_run_id="asdf"):
            EMAIL.run(flow_run_id="asdfasdf", name="TEST")

neither (context, nor the variable in the kwargs of the

run

function) actually change the flow_run_id when its output in the logger 😞 Do I need to wrap the flow_run_id in a state variable? If so whats the best way to do this? Finally, I'm running this flow over a dask cluster so idk if my issue here is affected by this issue: https://github.com/PrefectHQ/prefect/issues/2883

I created a project

vtk_flow

using the web GUI, and am attempting to register a flow:

$ python3 -m vtk_flow.flow
Traceback (most recent call last):
  File "/usr/lib64/python3.7/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/usr/lib64/python3.7/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/nih-3d-resource-workflows/vtk_flow/flow.py", line 82, in <module>
    f.register(project_name='vtk_flow')
  File "/prefect-env/lib64/python3.7/site-packages/prefect/core/flow.py", line 1583, in register
    no_url=no_url,
  File "/prefect-env/lib64/python3.7/site-packages/prefect/client/client.py", line 729, in register
    project_name, project_name
ValueError: Project vtk_flow not found. Run `client.create_project("vtk_flow")` to create it.

I thought Client was to talk to Prefect Cloud ?

Hi! I need a specific task to run if a previous task was completed + it is a certain day of the week. Is there a way to use the schedule component for filters?

Hey guys, is it "prefect legal" to inject context into

prefect.context

at runtime? More concretely, can I do something like this:

with Flow('test') as flow:
    do_stuff()

with prefect.context(foo='bar'):
    flow.run()

and if not what would be the correct way to inject context after defining a flow?

💃 Just arrived! Hello Prefect world

Hi Folks, is there a way to force the restart of a task from within the UI ? some tasks are getting stuck indefinitely in a pending state and I am trying to force the flow to attempt to run them ...

Hi All, I am looking at setting up Prefect on GCP and I did go through the medium post regarding that. I am looking for a server less solution to run the flows that will not have the overhead of setting up k8s. Something like Fargate on AWS ?

Hi, I have a quick question on cached tasks: will Prefect skip parts of the graph that are cached and don’t need to be computed so it can jump to only the tasks that do need computation?

💃 Just arrived!

Are you guys doing a cloud release cause I am unable to view my flow in Cloud

Hello everyone, I got some questions regarding the Prefect Agent. Firstly, how should I choose the number of agents, is there a way to calculate or common way to decide this number? Secondly, is a single agent enough to handle (maximum) 1000 querys per hour, the server is running with a local environment and using a DaskExecutor(a local Dask cluster)

Hello guys, I’d like to log some info in state handlers, such as flow version, flow run id, is that possible? How can I get them? thanks

Hi all, Is it possible to control how many mapped tasks are running at any one time? I have a memory issue, as my mapped task requires loading in some data, and no matter how many workers or threads I specify in the DaskExecutor, the tasks just keep spawning.

Hi all, question about latest server/ui ; I have prefect installed as a docker swarm, if pull the latest images ( tried "alpha" and "latest") the UI asks to create a tenant with command

prefect server create-default-tenant --name default

but from the shell ( inside the graphql container ) I got

Error: No such command 'create-default-tenant'.

prefect version

0.13.1+14.g303356902

EDIT: worked with master ( prefect v 0.13.2 ) great !

Happy Monday everyone! Are there any best practices for sharing “clients” between tasks? By clients I mean database clients/connections, API clients etc. I would assume there’s some care to be taken here due to the distributed nature of the tasks.

Hi, we made a small adaptation to the task runner to make it run a little faster locally so we could use prefect as a slick local DAG executor. When running one of these flows within a regular flow, we get all the logs for both flows when we really don’t want the inner flow logs. How would we shut off logs for the inner flow? Was thinking something like wrapping the inner flow like this:

with prefect.context(loglevel="ERROR"):
    inner_flow.run()

Hi all, I am looking to create a dynamic task DAG, which essentially calls a database for all task definitions and dynamically generates tasks. Here is a code snippet:

class DS242(ParkMobileFlow):

    @property
    def tasks(self):
        """Loads all tasks for data quality tests."""

        # Connect to ParkMobile client
        pm = ParkMobile()

        # Get Snowflake connection
        sf = pm.get("snowflake")
        sf.connect("sqlalchemy")

        # Call all tasks and task definitions
        records = sf.execute("""
            SELECT *
            FROM DEV_DB.DATA_QUALITY_TESTS.QUALITY_TASKS
            WHERE is_active = true;
        """)

        return records

    @property
    def flow(self):
        """Returns the Prefect Flow object."""
        return self.build()

    def build(self):
        """Builds the workflow."""
        tasks = []
        for task in self.tasks:
            tasks.append(
                DataQualityTask(
                    uuid=task.task_uuid,
                    name=task.task_name,
                    description=task.task_description,
                    source_query=task.source_query,
                    target_query=task.target_query,
                )
            )

        with Flow("DS242__data-quality-tests") as flow:
            for task in tasks:
                task()

        return flow

Now, I am trying to

register

this flow to the backend server, and am getting a serialization error, which is understandable. My question is, can I create a first

Task

, that queries the DB for all task definitions, and then dynamically creates new tasks to the

Flow

which performs all of the tasks? I can certainly build a custom Dockerfile that handles deployment of this, but would be awesome to connect to

prefect server

Hello all! I have a question about scaling to Kubernetes+Dask in GCP. I am trying to run a flow that uses pandas for data transformation that requires a large amount of memory, more than what can be allocated for a single node of our current cluster configuration. Here are some of the ideas, I had to fix the problem 1. I could parallelize my flow using Dask Dataframe but that seems to be a lot of effort. 2. We could vertically scale the nodes, but then we would pay for more compute that we need most of the time 3. Create a preemtible node pool where the flow can then be executed I feel like the 3rd point could be a good solution, but I am not sure how it could work to allocate a preemtible node pool at the start of a flow and then execute the flow on the new node pool since we have already deployed the job or dask cluster on the other nodes. So I guess my question is what is the best way to achieve what i'm trying to do without having to vertically scale the cluster? Thank you!

Good evening, I ran into the following problem, which I find it difficult to solve. My flow looks something like this: And here's an example: I am using localagent, and this flow will be executed sequentially for each file. What is the correct way to parallelize it? Use multiprocessing? I am more inclined to use the dask executor, I looked at several examples, but nowhere the flow was registered, but simply executed. What is the best solution to parallelize the calculations (the best option, I assume, that each node of the executor takes a separate file for itself)?

Good morning friends. A quick question on debugging with logs shown on the Prefect Cloud UI. I have some error logs buried among my info logs, but I’m having a hard time displaying them while keeping the info logs visible. I can filter for error logs, but then I want to see the logs surrounding those (i.e. the info logs) but I can’t seem to find a clever way of doing that in the Prefect Cloud UI. Any thoughts? Downloading the logs would also be an option but it seems it only downloads logs for the rendered page.

Hi everyone! Can I ask for help here?

I'm going crazy, I'm trying to pull a secret in a class task, and the type it returns is

<class 'prefect.configuration.Config'>

and I have no idea why. Did I do something wrong in the

config.toml

file??

I have created a python script. How can I run it in order to see the scheduling and flow status on the UI?

Hey guys, I'm trying to implement some basic task dependencies, but some of my tasks appear multiple times in the graph that

flow.visualize()

generates. In the graph that's generated I would like

Add new columns

to be dependent on

Extract customers

, while

Extract customers

should itself dependent on

Unnest dassport data

. In the code, how do I add the

Unnest passport data

dependency to the existing

.set_dependencies()

block without causing duplicates? Thx in advance

Hi all! I have a problem with flow again :)

How do I spec params in gql while trying to run a job? eg if I have a params called input_loc and submission_id, how do I define those args in gql?

mutation {
  create_flow_run(
    input: {
      flow_id: "dcbbfc94-5643-453c-941a-0efa8505df66"
      parameters: {"input_loc": "test"}
    }
  ) {
    id
  }
}

Hello, how can I reuse a flow? I have a function that takes in a list of items and build a flow using those items (I don't use prefect parameters currently). I want to deploy 2 versions of that flow (they will be different since the item list will be different) . I do change the name each time but it seems like they're both being considered as the same flow and replace each other.

Hey all, happy Tuesday!   I'm trying to import additional generated Python files and use them as part of my Prefect Workflow. Everything works fine locally, but when I push up the flow to a docker repo and try to run it using a Kubernetes agent, I'm getting the following error from cloudpickle:   ModuleNotFoundError: No module named 'test_pb2'   The Kubernetes agent spins up a job in our cluster, and this error comes from that job. It works fine if the flow doesn't reference any other custom modules.   Here's what I've tried so far, but it all results in the same error (note that the healthcheck.py passes just fine in all these cases when the flow is pushed to docker storage):   1.       Setting PYTHONPATH to point to test_pb2 and other files. If I pull the container, PYTHONPATH is correct; however, when the kubernetes agent spins up a job using that container, it doesn't appear like it's set anymore. Is it getting overridden somewhere?   2.       Turning the generated files into a package installable with pip and then installing it as part of the Dockerfile that creates the docker storage. If I pull the container, the package appears to be installed successfully. Still the same error when run by the kubernetes agent though.   Any ideas on other things I could try? Is there a preferred way to include other Python files as part of a prefect flow?   Any ideas/help is greatly appreciated!!