Hey all, hopefully this is isn't just noise. Have the need to integrate workflow management software into my multi-tenant SAAS software. Trying to figure out if Prefect is an okay match for my use case. I have a preset list of flows/tasks I want each tenant to be able to schedule / run independently. Each of these flows need to be parameterized or use context so I can specify things like "sftp connection info", "db connection info", which will differ per-tenant. Some of these flows must be limited to concurrency=1-per-tenant just to throw a wrench into things.

I'm trying to sign in to Prefect Cloud with GitHub. Worked yesterday but now getting this screen:

is there a way to dynamically assign resources to a given triggered flow

hi everyone, i thought you might enjoy the automatically generated name of our flow that just happened to fail we dont know what happend but my colleague @Greg Roche fixed it by reregistering it

Hello All, I want to one sqlalchemy engine through the flow, to avoid code duplicaton. My simple solution was to make a task which returns sqlalchemy engine object as result. It works when executing it locally, but returns error when executing it on server( cannot pickle local_object ). Is there a solution to this problem or is there a best practice how to share connection in a flow? Thanks!

Hello, I’m running into a memory issue with a mapped task that computes an array and then writes it to a DB. I think what is happening is that all the arrays (which can be quite large) are being held in memory. Since I’m writing these arrays to a DB I don’t need them to remain in memory after the mapped task has a Success state. I’m running these flows on a kubernetes container, and after X number of large arrays the memory request is exceeded and the pod is evicted. No other downstream tasks are dependent on the completion of the mapped tasks. Is there a way to tell Prefect to dump the task result cache after it has been mapped successfully? Pseudo code:

@task
def get_array_input(db_client, query):
	return db_client.query(query)


@task
def compute_array_write_to_db(input):
	array = requests.get_huge_array()
	db_client.write(array)

with Flow("generate-and-write-arrays",) as generate_and_write_arrays:
	lots_of_inputs = get_array_input(db_client, query)

	lots_of_inputs.map(compute_array_write_to_db)
    
# Crashes after certain amount of arrays 
# due to memory limit    
generate_and_write_arrays.run()

I am trying to thinking through a rapid pipeline without the overhead of Kubernetes or Docker images. When I register a flow how does prefect know what code to register. Could I do something like:

with Flow("Example") as flow:
    ....

def __main__():
   if os.getenv("production") == True:
       flow.schedule(my_scheduler)
    flow.register()

If I register a flow with that in the main would I be able to control if it has a schedule?

I managed to deploy to ECS a task that runs a local agent. The problem is that this agent is not able to run flows stored with Docker. I tried to chagne the agent on ECS to docker but it doesn't run. Is it possible to deploy and agent that runs docker on ECS? I don't want just to execute the flows on ECS. I need the agent itself to be running on ECS.

Has anyone seen this error before? It’s marked as Critical from the CloudHandler

Failed to write log with error: HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Max retries exceeded with url: /graphql (Caused by ProtocolError('Connection aborted.', BrokenPipeError(32, 'Broken pipe')))

In Prefect it says the task failed, but it still seems to be executing in the agent. There were logs before from the agent to the cloud, but no logs afterwards

Has anyone had an issue with Prefect not.

Prefect automatically gathers mapped results into a list if they are needed by a non-mapped task

I have two mapped tasks, one feeding the other, then a third non-mapped task which expects a list. The logger context doesn't appear to be getting passed so I cant get any insight. The UI is showing the child-mapped tasks and I have logged out their content but their results are not being gathered together as suggested in the docs. Any pointers to help would be amazing!

Prefect Team! A couple of time now I get disconnected from Prefect Cloud and sent back to login where I Select Google SignIn. A) The first time I land on some Okta page (See Screenshot) B) Then I have to click on the “Prefect” “Work” node. C) Back

<http://univeral.prefet.io|univeral.prefet.io>

D) Click on Google Again. E) Now I have Google’s prompt to select which account to use (Personal, Work, etc.) F) Then I’m logged in. Any idea? Can this be related to https://prefect-community.slack.com/archives/CL09KU1K7/p1611786977326500 ?

Got this error. Wondering what the largest entity is. I can trim the string down if needed, but just wondering.

Failed to write log with error: 413 Client Error: Request Entity Too Large for url: <https://api.prefect.io/graphql>

Hi Prefect, How can I mount a predefined Docker volume when using the Docker Agent? For example I have a volume named

"test-volume"

created with

docker volume create test-volume

If I want to mount the named docker volume

"test-volume"

with

prefect agent docker start --volume test-volume:$DOCKER_DIR ...

, the named volume is never mounted into the runtime started by the agent, a new volume seems to be created and mounted each time.

Hi, looking at Prefect vs Airflow. AWS' MWAA has a mechanism where you can upload a DAG to a designated s3 bucket and it automatically registers with Airflow. I've read a little about the file based flow idioms in prefect, but is there any mechanism to point at say github repo and automatically register the flows? I know it's just one step but if I were to have a lot of flows I'd love to be able to just store my flows in git, point prefect at my repo and have it manage the registrations based on my source

has anyone else had issues with the sample code hanging from the flow-to-flow documentation? https://docs.prefect.io/core/idioms/flow-to-flow.html

👋 I’m here! What’d I miss?

Is there a good way to parameterize an environment variable in a flow? Below is my user config. I'd love to be able to pass

mode

as a parameter instead of messing with it in the config and re-registering my flow.

mode = "hard_coded_mode"
[sql_server]
    server = server
    driver = driver
    dsn = "MYMSSQL"
    user = "${sql_server.${mode}.user}"
    database = "${sql_server.${mode}.database}"

    [sql_server.dev]
        user =  dev user
        database = "dev db

    [sql_server.hub]
        user = app user
        database = app db

    [sql_server.prod]
        user = prod user
        database = prod db

Sorry if this is a dumb question but what's the simplest way to run an agent in the background? I've just been using a screen session.

Hi there If I pass an object to a task (with the dask runtime). What exactly will happen to this object? I was thinking, it would be serialized and copied for every task. But it seems, that there will also be some logic runned. Is this process documented anywhere? Sure I can give more details, if needed.

Hello friends, our CI process is suddenly failing due to *errors in importing GCP / BigQuery / Google Core dependencie*s. We haven’t changed anything on our side so curious to know if anyone else is seeing this issue suddenly. Details are in the thread. Any help would be much appreciated!

Hello guys, we create flows to deal with big time series of GeoTIFF rasters (treated as xarrays). What do you suggest about passing massive data between tasks? Is better to store those files on S3, EFS temporarily or treat them as a regular file and return them from functions?

Hello guys! I'm using this example : https://docs.prefect.io/core/examples/functional_docker.html and I would like to know if it's possible to add a retry if python job fail - I saw how to do it with a task but not CreateContainer function

I’m using a Context Manager, but how can I mock the behavior of Prefect such that I can only test the setup function? I tried

@pytest.fixture
def mock_resource_manager(monkeypatch):
    mock = MagicMock(return_value=None)
    monkeypatch.setattr("prefect.tasks.core.resource_manager.ResourceManager.__call__", mock)
    return mock

But this gives weird errors down the road, as I also want to mock prefect.config/context for other purposes.

Has anyone got any thoughts on testing deployment process? I will be running my flows on k8s cluster but would like devs to be able to test out their flows locally before submitting a PR which will deploy the changed flow to the server - it's not completely clear to me what the best process is here so would be good to hear of other peoples approaches?

Hi, I am looking into building pipelines that involve GPU work on a very irregular basis that autoscaling is a must. In my case, dask is not really an option as it interferes with the multiprocessing of pytorchs dataloader and otherwise was rather unstable for my GPU workload. To make it simpler, I would like to start a flow that I know needs GPU on it's own instance and do the processing there. The nodes would need to spin up or down depending on current demand. Any pointers how to best achieve this? Currently, I see different possibilities, but not sure which one is best: • Spin up a new Agent on a GPU node before the flow is scheduled and then using tags • Somehow use KubernetesRun to request a GPU and let kubernetes handle the up and downscaling • Only use prefect to trigger a ECS job

Hi everyone, nice to join your community 🙂 We face an issue which seems to impact different google cloud platform projects, while it should not. In the thread-attached GCP log, we can see that it refers to both quetzal-omind as well as cartographie-cps projects. We do absolutly not understand why?! Our flow runs are pretty unpredictable since some days, most of the times they simply do not start, often they fail in a http requests loop, and sometimes it runs successfully! We suspect a wrong setup of our prefect agent, but we are pretty new to prefect dev, so we are very not sure of where to look for our issues. Maybe someone here will have some clues to help us 🤞 best!

How would one trigger a prefect flow from a github event? Let's say merging to master/main should trigger a new run.

Hello all, I am attempting to set up Slack Notifications for my flow and am following https://docs.prefect.io/core/advanced_tutorials/slack-notifications.html#installation-instructions. I have successfully installed the prefect app into my workspace and have been able to test the connection with success. However, I am unable to have it output information based on my actual flow. I receive the following error:

Exception raised while calling state handlers: ClientError('400 Client Error: Bad Request for url: <http://host.docker.internal:4200/graphql>\n\nThe following error messages were provided by the GraphQL server:\n\n GRAPHQL_VALIDATION_FAILED: Cannot query field "secret_value" on type "Query".\n\nThe GraphQL query was:\n\n query($name: String!) {\n secret_value(name: $name)\n }\n\nThe passed variables were:\n\n {"name": "SLACK_WEBHOOK_URL"}\n')

what am I doing wrong? I have @task(state_handlers=[slack_notifier]) set for the task I am testing out the notification for.

Is there a way to write a prefect flow to snapshot a DB?

Is there anyone who have run an

Apache Beam

cluster using Prefect. If so, like to get some input and lessons from your experience.