So I made a custom Secret implementation that pulls from google secret manager. The problem is that the dask-workers blow up because they can't seem to find the module. It's just a file in my demo directory. How is this supposed to work?

I tried having the workers use the same image as the scheduler which has the module, but that doesn't seem to cut it.

Is it possible to have a secret name come from a Parameter?

This is my attempt:

with Flow("mirror_sftp_to_gcs") as flow:
    password_secret_path = Parameter("password_secret_path")
    sftp_connection_info = Parameter("sftp_connection_info")
    password = GoogleSecretManagerSecret(password_secret_path)  # TODO: Cast function to convert to namedtuple
    host_path = Parameter('host_path')
    files = sftp_list_files(sftp_connection_info, password, host_path)
    downloaded_files = sftp_to_gcs.map(
        file_path=files,
        connection_info=unmapped(sftp_connection_info),
        password=unmapped(password),
        bucket=unmapped("snip")
    )

I get an error:

ValueError: Flow.run received the following unexpected parameters: password_secret_path

Very odd, the other parameters work, I am specifying them in a parameters kwarg in flow.run() in another file.

It appears that because the parameter is never used in a task, it doesn't ever expect it

How do I throttle a task being mapped?

I just started looking at the Cloud Scheduler yesterday and saw this on concurrency limits. https://docs.prefect.io/cloud/concepts/concurrency-limiting.html#task-concurrency-limiting Not sure what you do if you are running locally with Dask.

That's unfortunate it can't be rate-limited without cloud, I hope that feature comes back.

Within a single Dask cluster, you can use Dask resource limits to achieve a similar result. Global limiting requires some central broker of state.

Do you have a link on that by chance? I feel like thats something that should be handled by the executor if the underlying system supports it. I totally get you need to fund development but if you are thinking of open source as a strategy to onboard people to the commercial offering it's difficult to put together a POC without this 🙂

It is handled by the executor as the docs explain - https://docs.prefect.io/api/latest/engine/executors.html#daskexecutor

I meant that if the executor itself supports it, the Executor abstraction should facilitate it. Maybe I’m misunderstanding but it doesn’t seem like it does

Hi - Is this the correct way to handle a state in a flow that fails and skips all downstream task on failure. The result of the flow being a an empty list? Thanks again

Is there some way to pass a

Secret

to the slack notifier or look it up by name? Looking at the code it seems to read from the context, but it's not clear if I can actually set secrets in the context other than via the config file

Ah, I see there are Secrets and Secret tasks

seems like the notifier uses the former, so this probably won't work

Has anyone come across issues with Scheduled Flows not releasing the memory? We have an ETL setup to run every day that will just consume additional memory on each run. Are there any easy gotchas that can cause this behaviour before I begin digging into this?

hi all, I'm looking in to adopting prefect for DS workflows instead of trying to shoehorn them into airflow. does anyone have some experience with this? Has anyone found it slowing down workflows as opposed to something like kubeflow? Does prefect core have a web ui or only supports the dask ui?

Is there an example of how to setup a custom configuration file for Windows? It points to ~/Users/<username>. This is not a valid path for Windows users. I have tried setting the environment variables on load with no luck. I also set them in the windows environment variables and still no luck.

Hello. I’m playing around with the slack notification and trying to locally set the secret but keep getting the error that

ValueError: Local Secret "SLACK_WEBHOOK_URL" was not found.

I tried the suggestions mentioned here https://github.com/PrefectHQ/prefect/blob/b4c08b15b386107e2c1afb7fed35dca15b683cdd/src/prefect/client/secrets.py but wasn’t getting anything to work locally. Threading what I currently have

hey - is there a way to have a prefect task populate a global python

While we are queuing up questions this morning, is there a way to direct which Dask resource a task needs? The obvious example is GPU or CPU. In my case, I have one task that needs a lot of memory and I don't want to set the memory limit on all the workers to that level. I'd just like to have one worker with a high memory limit and send that task to that worker. https://distributed.dask.org/en/latest/resources.html#worker-resources

seems like formatting on the docs page on Constant tasks got borked https://docs.prefect.io/api/latest/tasks/constants.html

"No module named IPython" -- is that not included in the prefect module?

Hi all, Is there any way to store data between runs and don't execute some tasks if outputs already exists by analogue with luigi targets?

I found that I can specify

result_handler

for task, but it doesn't check that data already exists in a repeatable run

Hey guys, I'm using Prefect with Dask, using the built-in DaskExecutor. I see in the Dask gui that sometimes tasks 'linger'. They don't get cleaned up. Over time, this eats memory. Restarting the Dask Scheduler helps, but ideally I would see the tasks getting cleaned up by the system itself. Is this a known issue and is there a way to keep the memory clean? We're using dask==2.9.0 and prefect==0.7.3

Hi! I'm running flows using the k8s executor/agent in staging/production environment and local executor in development environment. For many workloads have the primary flow trigger work either: 1. If in staging/production environment - in a separate k8s job with a createNamespacedJob task followed by a custom waitForNamespacedJob task. 2. If in development environment - using docker createContainer, runContainer, waitOnContainer tasks Sometime these are created based on mapped inputs so work can be done in parallel - but obviously need some limits. In staging/production I can use tags for task concurrency limits which is all good 👍. How would you recommend I limit task concurrency when in development environment creating/running/waiting on containers?

I've been working to keep exception and log data in our environment instead of going to Prefect Cloud. Exceptions were pretty easily handled with a decorator to catch the exceptions, store them locally and then raise an exception that has the URI of the saved exception. I have a small app that pulls down the log from the cloud and builds a simple web page that lets you click to get the full exception from the local store. I wanted to do the same thing with logging. Log the message to a local file and update the message to Prefect Cloud with the URI of the locally stored log message. I can add the handler to log to a local file and I can update the formatter on the prefect logger to accomplish all of this when I have control of the application and call flow.run(). I'm having problems figuring out how to get the little stub of code that makes these change to the logging configuration to run with the agent picks up the flow from Prefect Cloud and runs the flow. Is there a place to put this code that the Agent will run it before the flow is kicked off? I guess as a last resort I can add a set task at the beginning of the flow to make this configuration change.