Hi Everyone, I am trying to come up with the best flow storage/deployment strategy for easy maintenance with a small team, but that is not going to be super painful to scale as we grow. We are running the Prefect Core server and have that configured properly, but I am struggling with the best approach to register flows and store them. I have pieced together a lot of different ideas from the documentation and slack, but I still think I am missing something. We have an ETL VM that we will be using as our host machine, it has Docker installed, and also hosts the Core UI. We use Azure for all of our infrastructure (VMs, SQL Server, Blob Storage, DevOps Repos, Container Registries) and are tied to that platform, but are open to using any other services available in Azure. We also have several internal python packages that we are actively developing (read, changing often) for interacting with our EHR and other systems. Looking at the available agents, the Docker Agent seems like it would be the best fit for us right now. However, I am struggling with the best approach to get our packages from our private repo and our flows into a docker image. Since we are using SQL Server, we have to customize our images to install pyodbc and the MSSQL ODBC drivers, the smallest image we have been able to develop has been ~600 MB. We are currently using a multi-stage build to clone our internal packages from Azure DevOps, then copying them to the image and setting the Python path. This works, but every time one of the packages changes we need to rebuild the image, push to the registry, and pull on the host machine; it feels extremely wasteful of cycles and storage. I was considering cloning the repositories to our host VM and using docker volumes to share the packages and flows, then we would not have to build the image every time we made a change. I was really excited to read the 0.12.5 release notes, I am hopeful the script storage functionality will benefit our use case, but I still need to explore it further. Does anyone have any suggestions on how to better architect this? I looked at https://docs.prefect.io/orchestration/execution/storage_options.html#non-docker-storage-for-containerized-environments but that seems to only be flow storage and I need to handle additional files that are updating very often. Thank you for any recommendations you may have!

Hello, I have another "quick" question branching from my previous issue. 1. If I make a change to my file say

first_flow.py

do I have to run

python first_flow.py

in order to get it to register the change? (

flow.register()

is in my py file) 2. I'm now seeing my flow in the UI but when I try to run it via the UI (as a one off run) it goes into "scheduled to run" but late and never runs (i just mark as cancel after 3mins).

import prefect
from prefect import task, Flow

@task
def hello_task():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hello, Cloud!")

flow = Flow("hello-flow", tasks=[hello_task])

# flow.run()
flow.register()

https://prefect-community.slack.com/archives/CL09KU1K7/p1595532174143800

Hi all, I am trying to solve the exact same issue: https://prefect-community.slack.com/archives/CL09KU1K7/p1585863863351300 The given solution is not working for me. I want to use Parameter in shell command:

from prefect import task, Flow, Parameter
from prefect.tasks.templates import StringFormatter
from prefect.tasks.shell import ShellTask


msg_task = StringFormatter(template='{my_string}')

shell_task = ShellTask()

with Flow("string-template") as flow:
    name = Parameter('name')

    msg_output = msg_task(my_string=name)
    shell_output_static = shell_task(command=f"echo Arthur")
    shell_output_dynamic = shell_task(command=f"echo {msg_output}")


flow_state = flow.run(name='Marvin')


print(flow_state.result[msg_output].result)
# Marvin

print(flow_state.result[shell_output_static].result)
# Arthur

print(flow_state.result[shell_output_dynamic].result)
# Command failed with exit code 2

Hey there, I’ve been working on my prefect workflows for a while now and I want to take it to production grade. To that end, I’d like to have my flow turn on a dask cluster, run for a while and then turn it off at the end (even if it fails). I was thinking about having tasks for that but since I’m configuring my flow to run on the cluster, these tasks would need the cluster running in order to operate. I guess there’s something here I’m missing regarding best practices for deployment, can someone here advise based on their own setup, or point me to the right place in the documentation? Thanks!

Hi. I have a scheduled, parameterized flow that missed a few flow runs. Is there a quick way to re-run a failed flow with the same parameters? My flow relies on the flow start date (context variable). I am using Prefect Cloud.

Hi, if I have certain run time conditions to check before defining some upstream tasks, can I still use the same task variable for dependencies?

When using the

DaskExecutor

is it safe to share the dask cluster between the execution of the tasks and direct use of dask? The use case here is I have a large number of small tasks I'd like to do in parallel, followed by some large joined data analysis. I'd like to gain the concurrency from running on a dask cluster in my flows, but at the same time I want to perform some data analysis using the dask

Dataframe

class and distribute this across a cluster as well. If I provision the cluster myself, is it safe to share the scheduler between the flow and the dataframe library? Or should I expect to need to provision my own dask cluster inside a task to run my Dataframe code on?

What are the capabilities of perfect?

Hey there, I’m having a flow with many mapped tasks (about 5k), running on a

DaskExecutor

. The flow fails due to high memory usage by the dask worker which is being killed again and again endlessly. Questions: 1. It seems that Dask/Prefect don’t serialize the completed tasks outputs when the worker gets to high memory usage. I’m not that proficient in Dask, I used Spark a lot and I know that it dumps to disk whenever memory is running low. How can I configure Dask to do the same (or is it a prefect thing?). Worth to mention that the prefect agent, dask scheduler and worker are all on the same machine. 2. Is there some way to have the flow fail in this case? It keeps restarting the worker but not failing, so I don’t get a message that this failure happens Thanks!

I have a kind of stupid question. I want to put my tasks into separate files to keep the code modular, but I want to assign the same state_handler to all the tasks. Is there a way to do this from the Flow?

Hello folks, I register a flow in this way:

import prefect
from prefect import task, Flow

@task
def hello_task():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hello, Cloud!")

flow = Flow("hello-flow", tasks=[hello_task])

flow.register(labels=['public'])

But the label of a flow run is:

"type": "LocalEnvironment",
  "labels": [
     "public",
     "AppledeMacBook-Pro.local"
  ],

Is there a way to get rid of label “AppledeMacBook-Pro.local” for it impacts the agent to pick up the flow run? thanks

Hi everyone, I was wondering what configuration is required to not see the below error. So, by default the

PREFECT_SERVER__GRAPHQL_URL

is set to http://localhost:4200/graphql but I want to change it some other URL. How can I achieve this?

Hello guys! I am new to Prefect. When I am running Prefect Core locally on my mac I just cannot start running the agent. I followed the step as the document describes and just got the information: prefect.utilities.exceptions.AuthorizationError: No agent API token provided. I don't know where to find the API token for my local server. Maybe I missed some information from the Doc ? Please help me make it up and run, thank you !

Problem solved thanks to another thread with the same problem :)

Dear all, I am currently trying to use AWS ECR as container registry, after having successfully experimented with dockerhub. What is the best way to authenticate via prefect to AWS ECR? 🙂

Hi everyone, hope you all had a lovely weekend. Would anyone have some examples of deploying flows on a CI server? I'm trying to get a better understanding of when to call storage.build() and how we can abstract that a bit for our analysts. Thanks!

Does anyone else experience the issue with the parameter field breaking up in the run view (in Cloud UI)? Firefox on Win/Chrome on Mac

Hi, I'm running into problems running MySQL commands from prefect. I've tried pymysql, mysqlconnector, and sql alchemy. I can't figure out if this is a mysql problem or a seeting I'm missing or a Prefect problem. The code runs and ends fine in a jupyter notebook. However, when prefect runs it it never gets the signal that the command ends. I even tried packaging it up in a stored procedure and everything ran perfectly fine. Prefect just never got the response that that it completed and I can't figure out why. Any, help would be appreciated. I also have other sql commands in other tasks that run just fine. I've attached the code.

I'm also a little new to using the built in tasks. I want to try the mysqlexecute task but how do i pass in upstream tasks to it?

Hi, I cannot seem to get my flow to run on a LocalDaskEnvironment from the cloud UI. When I use

flow.run()

locally it seems to work. Here is the code snippet I am using for registering my flow. I am using

Docker

storage.

if args.base_image:
        storage = Docker(
                base_image=args.base_image,
                registry_url=args.registry_url,
                local_image=args.local_image)
        for flow in flows:
            path = storage.add_flow(flow)
            print(f"storing flow {flow.name} at {path} in the image.")
        storage = storage.build()
        for flow in flows:
            flow.storage = storage
            flow.environment = LocalEnvironment(
                    executor=LocalDaskExecutor(),
                    labels=args.labels)
            flow.register(project_name=args.project, build=False)

Am I doing something wrong? I am on the developer version of the cloud backend. EDIT: To be more precise, when I run

flow.run()

I see concurrency on the mapped tasks, but not when using the cloud API (even for a single flow).

Hi everyone, a quick question. What is the relation between the Docker container and Dask? I'm struggling to make the link between the two. To my knowledge Dask is not aware of containers? (I've never used Dask but used k8s/docker extensively so just looking to find the link between how the container is run against Dask)

Anyone have experience / recs for doing Prefect <> GCP serverless (any form)? This is part of some citizen journalism web scraping efforts (twitter, fb, ...), so will be needing to vet & recycle IPs, and thus expect nodes to be somewhat short-lived (1min-1mo), and with many native serverless, 15min anyways

(We already have prefect for our normal GPU jobs on our stable cluster)

Using Prefect cloud, let's say I want to set flow SLAs (to be explicit: I want to be able to set cloud hooks, or something similar, when a flow does not complete a certain amount of time after its scheduled start). Is there a good integrated way to do this integrated with cloud? I see some copy about workflow SLAs on the website, so I assume something like this exists, but I don't see any other references in the docs/github/slack.

Can someone tell me what the little yellow exclamation point next to the "SCHEDULE" toggle button means?

Hi everyone, there a way to stop storing task results?

Hi. Is it possible to define a timeout on a flow run's (not task!) runtime? Because sometimes a flow is stalled. The flow is in state RUNNING but every one of the flows' tasks is at state PENDING. I have a prefect-server setup with a dask scheduler and some dask worker doing the heavy lifting. All components running in containers but prefect is NOT spawning containers (static server setup). Unfortunately I had to set a lifetime (=auto-restart) for dask workers (because after maybe some hundeds finished tasks the dask worker's cpu AND ram consumption is increasing -- even when no flow is running the cpu is running at >60%!). The RUNNING flow state with PENDING tasks seems to be persisted in the DB but is never updated, checked, deleted. After a complete restart of all components (apollo, dask, agent, etc.) the flow run(s) are still in the same state but nothing happens. This stalled flow run blocks every other task because its started by a scheduler. Could it be helpful to define a ResultHandler to allow the agent to pickup the flow run's crappy state and maybe continue or abort the broken flow run? I'm using a NFS to share data between dask worker and agent (mounted under

$PREFECT__HOME_DIR

. I see my pickled flows and a lot of pickled results. But Prefect seems not to be aware of that. The last log line is btw

Flow 'session-creation-watchdog': Handling state change from Scheduled to Running

. In a successful run, the next log line would be the start of my first task. I don't know where to start to debug that. What can I do to better understand the problem? Is there a timeout or can I force a "PING" to the flow to identify the real state?

Hi. How can we run an existing flow version in S3 from the Prefect UI? Any directions?

Hello Prefect Community! Are there any example repos of complete projects using Prefect server? I feel fairly confident in my understanding of the component pieces, but I'm having trouble seeing the whole thing as a finished product.

Hi. Newbie here. Was checking fitment of prefect for a etl project. Can we trigger prefect flow based on condition. Say, when a file arrives ? (like a file polling service)