Hello, I am new to prefect and have executed parts of the tutorial of basic features applied to our environment (Exasol Data Warehouse). I managed to execute arbitrary SQL statements on the Exasol Data Warehouse via a Prefect flow composed of tasks executing sql functions on Exasol in a specific sequence. So far I am running everything locally on my laptop. I have a few questions mainly on deployment for production environments, perhaps you can help: 1. Regarding deployment on a production environment, I would like an installation on on a docker host. However I do not want to install anything on the docker host itself (things like conda, prefect package, etc.) - rather I want everything encapsulated within separate docker containers. I ran across https://github.com/flavienbwk/prefect-docker-compose - which explains the deployment via custom docker compose in much more detail but it seems that on the docker host, prefect has to be installed in the OS as well. Are there other similar resources or alternatives? 2. Is there a open source solution for the Prefect UI to add user administration and user rights? 3. Is it possible to execute a single Prefect Task within a docker container? 4. Is it possible to execute a single Prefect Task with Kubernetes? 5. does Prefect or some other company provide support for on premise deployments?

Hello everyone, I had a general question about Prefect's FlowRun as shown here: https://docs.prefect.io/core/idioms/flow-to-flow.html and I wanted to make sure that I was understanding and that this made sense for our use case. We are considering the idea of having a "monolithic program" that has commonly used functions so we can just set parameter values on other programs where things differ. For example, we have a commonly used connect_to_db method that all our software shares. With a FlowRun would it be possible for us to register a flow that does the database connection and returns the information back to the dependent flow? As in Flow1 calls connect_to_db flow, returns connection information back to Flow1 and proceeds on with the rest of the tasks.

Howdy. We have a flow we are running that uses the core server to manage it. The flow hits a certain task and then just stalls out. It does not write any of the logs, it ignores the timeout set in the

@task

settings, and just sits waiting for an hour until the heartbeat dies. We can run it on the same worker server manually with no problems at all. There doesn’t seem to be any logs indicating it is out of memory. This seems pretty squarely on the prefect side but I am lost how to debug it since only the server has the issue.

Process 27956: /opt/prefect/.venv/bin/python3.8 -m prefect heartbeat flow-run -i 14303ee3-e96a-49c9-87e6-6577fcace7d3
Python v3.8.0 (/usr/bin/python3.8)
Thread 27956 (idle): "MainThread"
    flow_run (prefect/cli/heartbeat.py:98)
    invoke (click/core.py:610)
    invoke (click/core.py:1066)
    invoke (click/core.py:1259)
    invoke (click/core.py:1259)
    main (click/core.py:782)
    __call__ (click/core.py:829)
    <module> (prefect/__main__.py:4)
    _run_code (runpy.py:85)
    _run_module_as_main (runpy.py:192)

Hi All, does the flow object has task state information stored as well. I want to use state handler at flow level but would like to send task level msg if there were any task failures in the flow.

Hi everyone. I'm trying to execute a flow on kubernetes with a Dask Executor:

flow.storage = Docker(registry_url="702*****<http://487.dkr.ecr.eu-central-1.amazonaws.com|487.dkr.ecr.eu-central-1.amazonaws.com>", image_tag='latest', python_dependencies=["dask_kubernetes"])
flow.run_config = KubernetesRun()
flow.executor = DaskExecutor(
    cluster_class=lambda: KubeCluster(make_pod_spec(image=prefect.context.image)),
    adapt_kwargs={"minimum": 3, "maximum": 5},
)

flow.register(project_name="AWS_Prefect_Kubernetes")

But I get the following error:

Unexpected error: AttributeError("can't set attribute")

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/flow_runner.py", line 418, in get_flow_run_state

with self.check_for_cancellation(), executor.start():

File "/usr/local/lib/python3.8/contextlib.py", line 113, in __enter__

return next(self.gen)

File "/usr/local/lib/python3.8/site-packages/prefect/executors/dask.py", line 211, in start

with self.cluster_class(**self.cluster_kwargs) as cluster:  # type: ignore

File "C:/Users/giova/Git_Projects/PentaTechnology/earthalytics-engine/dask-flow-new.py", line 48, in <lambda>

File "/usr/local/lib/python3.8/site-packages/dask_kubernetes/core.py", line 414, in __init__

super().__init__(**self.kwargs)

File "/usr/local/lib/python3.8/site-packages/distributed/deploy/spec.py", line 274, in __init__

super().__init__(

File "/usr/local/lib/python3.8/site-packages/distributed/deploy/cluster.py", line 69, in __init__

self.name = str(uuid.uuid4())[:8]

AttributeError: can't set attribute

Any advice? It seems it's not be able to set the name of the pod into the cluster. I'm using Prefect 0.14.6.

Hi, we are using Prefect Core and running into an issue with memory handling. We are running in a dask distributed environment. We are using the mapped function (over 400 simulations) but I have replaced this entire function with a sleep(20) i.e. not doing anything but have kept the inputs as is. We are finding that each mapped prefect task is hanging onto the inputs for the entire flow duration. The dask workers (capped at maximum of 10) slowly increase as more simulations are run on the same pod (without any output). We've minimised the problem down to passing in any input parameters (whether used or not). Where the memory allocation is manageable if no input parameters are passed (with only a sleep function within) and then problematic when we add the input parameters back in (still with on the sleep function). The Prefect Documents state that input caching occurs automatically. Is there a way to specify clearing that cache and freeing the memory after the task has successfully executed? Thanks so much!

Hi folks, one of our flow runs just failed due to a task run failing with the following message:

Task timed out during execution.

- this is quite unexpected given we don’t have any timeout set on the prefect task, has anyone else encountered this behavior before ?

Hi Community, We had Prefect Server UI running for weeks on a GKE cluster using the helm chart. Recently, I noticed the UI wasn't displaying properly and just shows a blank screen. I opened up the Chrome developer tools in the browser and 4 JS errors popped up. Any insight on this?

Hi, has anyone tried creating dependent flows in Prefect. I tried to follow the example mentioned on prefect https://docs.prefect.io/api/latest/tasks/prefect.html#startflowrun but I am getting error "ImportError: cannot import name 'StartFlowRun'

i'm trying to use a prefect secret in the most basic way and am having some difficulty...my call

Secret("SECRET_KEY").get()

fails when I try to register my flow:

ValueError: Local Secret "SECRET_KEY" was not found.

my backend is set to

cloud

, and I do not have

use_local_secrets = true

set in my config.toml, so I'm not sure why prefect is trying to actually retrieve a local secret there

Hi all, I was wondering if there is a nice way of changing the UI port? The instructions in prefect server dont apply as Im attempting to modify the docker-compose, but the below configuration for the UI doesnt seem to work, same when I change the env variable to

PREFECT_SERVER__UI_HOST_PORT

. Our port 8080 is already in use by another service. Any ideas?

Are task

run

methods allowed to return arbitrary objects?

class ReturnObject:
  def__init__(self, value: str = "hello", list: List = ["world"]):
    self.value = value
    self.list = list

class MyTask(Task):
  def run(self) -> ReturnObject:
    return ReturnObject("some value", ["a", "b", "c"])

task = MyTask()

with Flow() as flow:
  result = task()
  some_other_task(param1=result.value, param2=result.list)

If I try to run this code, I’ll get an error like

AttributeError: 'MyTask' object has no attribute 'value'

So I am not quite getting something, is it possible to use local storage with multiple local agents as long as the flows are all stored in the same location on all agent servers?

When I register my flow, it's label is automatically set to the hostname of the one agent I registered it from

Hi all, anoter query from me: How do I tell prefect server where the server actually is. I now have a docker container up in the cloud with the 4200 port exposed, and I need to connect to it and run the command to create a default tenant. However, I cannot find any documentation in the prefect.cli for this and the help menu for prefect backend just tells me there are options, but gives no details on what they are (screenshot attached). How can I say something along the lines of

prefect backend server --address=<http://something:4200/graphql>

?

Hi All, I am trying to use Prefect caching at the task which works perfectly fine but I am having couple of issues: 1. When the same flow is run almost at the similar time which contains two task A & B. For caching to come effect the first flow has to run task A (then it cache it) and similarly when it completes task B (it would cache it). The issue comes when the same flow is run almost similar time when the tasks are not cached as the first trigger Flow might be still be processing task A and suddenly second Flow sees that task A is not cached and it would start even processing the same task A. 2. We do have some complex workflow that we are moving away from Celery without much rewriting. There are times when our two flows are different but they share some common task and I wanted to make sure that when these two different flow run at the same time the common task might end up running twice which is bit time consuming and cpu intensive. I am not sure how best to proceed with the above two issues.

Hello all, this should be a fairly simple question to answer, I just am having trouble locating the information in the documentation. I am wanting to enable checkpointing globally and I see that there is a way to do it temporarily via exporting the env variable, but I was hoping to know what I needed to add to the config.toml file for it to be a permanent feature. Thank you in advance!

One thing I am also not quite clear on, is what the real benefit of stored_as_script is? What does it buy you that pickling doesn't?

I may have overlooked that in the docs, but I feel like i've been over them 5 times already

Hi, I am returning multiple values from a task and using task decorator with nout option(@task(nout=2)) but I am getting TypeError: __init__() got an unexpected keyword argument 'nout'. Am I doing anything wrong?

Hi Prefect community! We’re using prefect cloud as backend, Docker Storage which uses a local Dockerfile and the prefect docker agent which uses the local image created by that local Dockerfile. We’ve created an AWS ECR Repository to store the image and I’d like to use that instead of the local one. I had the impression that I could do that by removing Docker Storage’s

dockerfile

argument and adding a

registry_url=MY_ECR_URL

param, but that doesn’t work and if I keep both params it tries to push the image to the ECR repo which I don’t want. What’s the right setup for this case?

Hi, I have a use case which might be not really in the scope of prefect, just wanted to raise some awareness around it, maybe it's something that could be considered :) We are using prefect as kind of a computation platform, just a place where you can run certain things without having to set them up locally, so using the manual trigger a lot. I would like to trigger some flows through the API, but as far as I can tell, there's no simple way to get a response back about the RESULT (for example where the output was written to) of the flow. To do so, I would have to trigger the flow, poll the state of the flow until finished, get the result of the last task of the flow, and then return that. Are there any other options that I did not see? I am aware that this probably is not the main focus of prefect, but other option would be to switch to something like celery for some tasks that need a response. Thanks

Hi team! I was wondering if anyone has any systems of productionising flows that they are happy with. We’re currently looking at systems within the Gitlab CICD that on push to dev/main will clone out a repository of flows and register each of them (uploading into an S3 bucket). I believe we could also use the Github/Gitlab storage option (question - does this support branching?), but we’d still need to clone it out so we can register each of the flow files. Has anyone got a nice setup that allows for rapid iteration and automatic flow registration that theyd be happy to talk about?

Hi to all, please contact JetBrains team: https://twitter.com/asm0di0/status/1359224228269154305?s=19

Hey all! Having issues getting the DockerAgent attempting to do some DockerRun flows not being able to access the registry. We have a set of images in an Amazon ECR location, but I cant see any registry setting options in the DockerRun or DockerAgent (and having the

AWS_ACCESS_KEY_ID

and

AWS_SECRET_ACCESS_KEY

set doesnt seem to be what prefect is looking for), and it doesnt use them to authenticate. Does prefect only support public images?

Hello! Can anyone from Prefect comment on Cloud's SOC2 certification? I'm out of my depth on this one but our devops team is asking.

Had a question on just the general approach to automate tasks that need to run on specific machines in the environment? Like I would like to get my database backups scheduled through Prefect, so should I have a Prefect agent installed on my backup server so I am able to kick off that task with shell? This is in addition to my other jobs which are more client-specific ETL tasks able to run from just about anywhere behind my firewall without issue

Having ye old issue of “Flows are submitted and an agent sees them, but they get stuck in ‘submitted for execution’“. I am wondering if anyone knows how to debug this. I know that: • the labels on the task and the agent match • the docker agent has pulled the image, started a container and “completed flow run submission” from the logs • Using either the Local executor or LocalDaskExecutor does not change the fact that: • Flows never progress from the “submitted for execution” state Is there a way of trying to look inside an agent to figure out what is not working?

I have had this issue with 2 flows now, getting a flow state of

Failed

on the UI, while every task and the flow itself is working as expected.. what can be causing this?

[10 February 2021 12:06pm]: Unexpected error: TypeError('Could not serialize object of type Success.\nTraceback (most recent call last):\n File "/usr/local/lib/python3.7/site-packages/distributed/protocol/pickle.py", line 49, in dumps\n result = pickle.dumps(x, **dump_kwargs)...

👋 I have a question about Result/Target usage, I would like the target template to be based on the flow parameters - how would I do that? ``````