Hi, I'm extending MySQL Execute and Fetch tasks to override init method, while 'run' remains the same. I'm initializing tasks outsize of flow() and then just call them expecting that second task is dependent on first one. Definitely doing something wrong. get_time (line 5) is just initializing a task, not getting actual output from select query.

fetch_task = TestMySQLFetchOneValue(connection_name='...', query="select now(6) as time_column", name="Fetch Time")
execute_task = TestMySQLExecute(connection_name='...', name="Write Time")

with Flow("Test Flow") as flow:
    get_time = fetch_task()  #fetch_task().run() works
    execute = execute_task(query="update test_run set run_time = '" + str(get_time.get('time_column')) + "'")

flow.run()

Hello, I have successfully deployed a Kubernetes Prefect Agent on Azure Kubernetes Cluster. I am trying to run a simple flow that utilizes a LocalDaskExecutor on the AKS Virtual Nodes. For this, I am using a custom job template for the pod, because it needs some customized node selectors and tolerations that Azure publishes. the following is snippet of my job_template:

job_template={
            "apiVersion": "batch/v1",   
            "kind": "Job",
            "spec": {
                "template": {
                    "metadata": {
                        "labels": {
                            "execution-model": "serverless"
                        }
                    },
                    "spec": {
                        "containers": [
                            {
                                "name": "flow"
                            }
                        ],
                        "nodeSelector": {
                            "execution-model": "serverless"
                        },
                        "tolerations": [
                            {
                                "key": "<http://virtual-kubelet.io/provider|virtual-kubelet.io/provider>",
                                "operator": "Exists"
                            }
                        ]
                    }
                }
            }

However the flow fails. When i ran kubectl get events. I notice the following output:

Warning   ProviderCreateFailed   pod/prefect-job-XXXXX-XXXXX   ACI does not support providing args without specifying the command. Please supply both command and args to the pod spec.

Just some more information - I also ran the same flow successfully on a alternate deployment on AWS EKS Fargate, using an AWS Kubernetes Agent. Any guidance is really appreciated :)

Hey guys! Quick question about the DockerRun config. I am running a docker agent on an EC2 instance and have specified a custom image that I have added my flow dependencies to. But keep getting an 404 image not found error(no such image: repo/image_name:tag). This is the same even from prefect core images for a simple test flow. Is there something I could have missed in the agent config?

Morning, I am testing a crash recovery scenario with prefect 2. My workflow has a few steps, all of which do lengthy sleeps. I have an agent running on a compute engine instance (wont be our final deployment architecture but is convenient for testing). I suspended the compute engine instance mid flow run. The log in the prefect UI indicates "Crash detected!" however the TaskRun still shows as Running. After restarting the agent it looks like there is no automatic crash recovery, so in this scenario we would need to setup a flow timeout? Is there any way to resubmit a TaskRun, and do all the agents operate in this way?

I'm looking at a scenario where a flow could be triggered a number of times simultaneously (via the api) - I think the extreme case is around 300x within a few seconds. I don't mind the tasks queuing up, but is there a point at which something will stop working from the Prefect side of things? I need to make sure the flows will indeed run - sooner or later.

Hi guys, I'm currently facing an issue with Parameter, how can i get the value of a Parameter ?

Hey, Congrats for releasing prefect 2.0 (beta). We currently were in the middle of using prefect 1.0 and none of our pipelines were production ready so we decided to move on and adopt Prefect 2.0 orion at first. At our pipeline we were using <task>.run to call a task from other task (not sure if that was kind of a workaround), but now this feature seems to be lost. Is there any workaround for doing that on prefect 2.0 ? (calling a task from other task) Are there any guidelines for moving from prefect 1.0 to prefect 2.0? Thanks!

Hi. My scenario is to create a prefect flow and then run it using an API call. I am able to run any already created flow using POST calls (screenshot1 attached). When the POST call is invoked, the function prefect_flow is called. But I am not getting idea on how to create flows by POST calls. (screenshot2 attached for create flow code without any reference to API). I tried to encapsulate the whole task and flow inside one function (temp_prefect_run, line#9), and then calling that function for POST call, but I am getting internal server error. When checking the detailed logs: if I am doing f.register(project_name), the error says ->

requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: / (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x000002A9EA91E100>: Failed to establish a new connection: [WinError 10061] No connection could be made because the target machine actively refused it'))

My prefect server is not at localhost. if I am doing client.register(flow, project_name), the error says ->

raise ValueError("This flow has no storage to build")
ValueError: This flow has no storage to build

What is the best way to run POST calls which can create a new flow?

Hello, is there an example of how I add s3 file as an attachment to prefect email task , I looked at the code and the current attachments param only support local uri , TIA

I have implemented distributed dask with the prefect. The task is working perfectly fine with LocalDaskExecutor but when I tried to run it in a Distributed Dask it has given an Error. I have a docker-compose of dask with 1 schdular and multiple workers. Every thing is up and works perfectly in case of local.

Any One Have idea then kindly help!

Error which i am getting : | Unexpected error occured in FlowRunner: ModuleNotFoundError("No module named 'prefect'")

Hi everyone!

I’m using the prefect master branch docker image and when prefect spins up, it’s throwing a json decode error. I know that’s general, but has anyone encountered this?

Tracing it through graphql to hasura currently

Hi, when trying to install prefect orion in kubernetes can I specify which namespace I want the service to be applied? Also, when I run the

prefect orion kubernetes-manifest | kubectl apply -f -

I got an error pulling the container because it didn't find docker.io/prefecthq/prefect:dev-python3.8. I've edited the deployment manifest to use docker.io/prefecthq/prefect:latest-python3.8 instead but I'm guessing that this is not the correct container either since I'm now getting the following error in the pod logs:

Usage: prefect [OPTIONS] COMMAND [ARGS]...

Try 'prefect -h' for help.

Error: No such command 'orion'.

Hi! has anyone faced this error when starting Orion? the .db file does not exist, it is new.

Hi, I am not able to locate config.toml in my macbook, I only see auth.toml and backend.toml. please advise

Hi is there a way to roll back a flow to a previous version?

hi prefect-community! is there a good way to use flow Parameters when configuring tasks before runtime -- for example, i'd like to pass company_id/company_name as parameters to my flow, then use those values to configure where to cache task outputs:

Co-worker put this issue request in: https://github.com/PrefectHQ/ui/issues/1261 Is this a current feature we’re missing?

Hi, I'm experimenting with best practices according to https://docs.prefect.io/core/concepts/best-practices.html#writing-prefect-code, and don't get why the following code doesn't build proper execution flow. It should be single top to bottom DAG, but instead I got some mess. Task are initialized outside on Flow imperatively; inside Flow tasks are executed, most of them have no output and dependency is set via .set_upstream.

with Flow("Test ETL Flow") as flow:
    start_task_result = start_task()
    truncate_task_result = truncate_task()
    fetch_task_result = fetch_task()
    update_template_result = update_template_task(variables=fetch_task_result)
    merge_task_result = merge_task()
    complete_task_result = complete_task()

    truncate_task.set_upstream(start_task_result)
    fetch_task.set_upstream(truncate_task_result)
    merge_task.set_upstream(update_template_result)
    complete_task.set_upstream(merge_task_result)

flow.visualize()

Are there are any examples of passing environment variables into the Docker storage image with the flow script?

In my imagination, I'd like to have os.getenv calls at the very top of my flow.py that can pull in private vars in production for credentials and such.

Working in Google Cloud and scratching my head a bit

Hello! I’m running a pretty extensive pipeline on a large EC2 machine (we’re still trying to get setup to port it to prefect cloud), and have some bottleneck tasks in the middle that collect the results of a bunch of previous mapped tasks and do some calculations across the whole dataset. The bottleneck tasks don’t manage to succeed, but they don’t explicitly “fail”, either, earlier tasks just re-run (collecting the checkpointed results) and then the bottleneck task starts again, but again does not finish… I can’t figure out what the problem is, I can run through the task outside of prefect if I load all the data, so I’m not sure if it’s a silent memory issue on the worker running the large task, or some time-out issue with the scheduler, or something else I’m not thinking of. Has anyone else run into similar behavior and have a suggestion for what to do to diagnose and/or fix the problem?

Will it be crazy if I think I can use Prefect as CI/CD tool 😁 , Prefect has most of the thing out of a CI/CD pipeline, graph dependencies, notification, artifact, task libraries

Hi , Inside task I am using a third-party library or you can say other module python script and bound the volume to the agent. Still got an issue Module not found error. Is there any way to solve this or do tasks have some restriction that they can not call library out of scope? In simple, How can we use external modules?

Hi, I would like to get flow_run_id from a flow created via create_flow_run(), how would I be able to do that? For context, I’m trying to create a FlowRunView of a child flow created by create_flow_run().

Hi , i am runing the tutorial to get the first heloworld in perfect.io , but i get this error message in the import statement >>> from perfect import task, Flow Traceback (most recent call last): File "<pyshell#2>", line 1, in <module> from perfect import task, Flow ImportError: cannot import name 'task' from '<unknown module name>' (unknown location)