prefect-community #prefect-community

Hi The flow running on ECSTask is executed periodically. However, yesterday’s flow remains pending without starting for some reason. Is there a way to monitor whether or not an auto-scheduled flow has been executed? Right now, I am notifying Slack after execution, but this is only when the execution is done. . thank you

flapili

02/28/2023, 10:28 AM

Hi, I have an EC2 VM with orion (renamed API few days ago if I understood correctly) and an EC2 VM with an agent what would you recommend/any ressources/docs to handle scalability & installing of (python) dependancies ? for now I'm login via ssh to the agent, git fetch, git checkout the poetry lock /pyproject.toml then installing dependancies I don't have time and competence to use maintain kube for now, is managed kube suitable for my small usecase ?

Ton Steijvers

02/28/2023, 10:41 AM

Hello. When I call a task and pass a Block instance, it looks as if the task receives a copy of the block instance, not the actual reference:

Copy code

@task
def mytask(obj):
  get_run_logger().info(f"in mytask: {id(obj)}")  # task shows different id than flow
  with obj.get_connection() as connection:
    connection.execute_string("SELECT 1;")

@flow
def myflow():
  obj = SnowflakeConnector.load(block_name)
  get_run_logger().info(f"in flow: {id(obj)}")
  mytask(obj)
  mytask.submit(obj)

Is this behaviour by design that I can rely upon? Are there cases where I would get the actual reference instead of a copy? _After having used the (SnowflakeConnector) object in the task, its underlying connection might have been closed. If subsequent tasks would receive the same instance, this would result in a "Connection already closed" exception. Getting a copy ensures that I can safely call get_connection in every task._

jpuris

02/28/2023, 2:09 PM

Heya!! This might be stupid question 😅 I have a flow with multiple tasks.. in something like airflow I'd need to explicitly define the dependencies of tasks in a dag (flow) i.e.

Copy code

task_1 >> task_2
task_2 >> task_3
task_2 >> task_4
...

Now prefect determines this implicitly by the "parameter" direction from task to task i.e.

Copy code

result_1 = task_1
result_2 = task_2(result_1)
result_3 = task_3(result_2)
result_4 = task_4(result_2)

Which is fine, no problems there.. but what do we do when there is task 5, that does not require any result from the tasks it is dependent on? For example task 5 absolutely must run, if task 4 is done, but has no other relationship with it, like

Copy code

result_1 = task_1
result_2 = task_2(result_1)
result_3 = task_3(result_2)
result_4 = task_4(result_2)
result_5 = task_5()

With sequential task runner (default), this is OK, but if I were to use concurrent one (I want to have the tasks 3 and 4 run at the same time, but 5 must not run before task 4 is done?

✅ 1

Thet Naing

02/28/2023, 2:53 PM

Hi all! My team is considering a switch to Google Kubernetes Engine for our Prefect infrastructure. Does anyone have recipes or examples of deploying flows with GKE clusters in Prefect 2.x? Any help would be greatly appreciated.

Dave D

02/28/2023, 4:48 PM

Does Amazon AppFlow integrate with Prefect?

jpuris

02/28/2023, 6:39 PM

Heya! Is there a way to fail a flow intentionally without printing the stack trace of the failure?

Copy code

if not is_success:
        logger.error(error_msg)
        return Failed(message="I've failed because ...!")

Copy code

19:31:34.931 | ERROR   | Flow run 'sociable-mantis' - Traceback (most recent call last):
  ... actually relevant stack trace from "error_msg" ...

19:31:35.161 | ERROR   | Flow run 'sociable-mantis' - Finished in state Failed('I've failed because ...!')
Traceback (most recent call last):
<... a massive stack trace on raising a failure from prefect :(>
    raise await get_state_exception(state)
prefect.exceptions.FailedRun: I've failed because ...!

The last stack trace is really unnecessary. I'd only want to output information that the flow has ended in failure state 😞

✅ 1

Tony Alfonse

02/28/2023, 8:37 PM

Hey community, is there a way to execute a deployment/flow from PowerShell?

Anand Soleti

02/28/2023, 10:17 PM

Hello Team - I'm trying to use prefect gcp collection in my python code and having issue as below.Anyone encountered same. Warning! Failed to load collection 'prefect_sqlalchemy': ImportError: cannot import name 'CredentialsBlock' from 'prefect.blocks.abstract' (/home/anand/anaconda3/envs/zoomcamp/lib/python3.9/site-packages/prefect/blocks/abstract.py) Warning! Failed to load collection 'prefect_gcp': ImportError: cannot import name 'CredentialsBlock' from 'prefect.blocks.abstract' (/home/anand/anaconda3/envs/zoomcamp/lib/python3.9/site-packages/prefect/blocks/abstract.py)

Maikel Penz

03/01/2023, 2:01 AM

Hey team. I'm getting an error to deploy the agent on Kubernetes. This is what I am running

Copy code

export PREFECT_API_URL="<https://api.prefect.cloud/api/accounts/><ACCOUNT-ID>/workspaces/<WORKSPACE-ID>"
export PREFECT_API_KEY="<API-KEY>"
prefect kubernetes manifest agent -q <MY-QUEUE> | kubectl apply --namespace=default -f -

And the error..

Copy code

| _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|


Agent started! Looking for work from queue(s): <MY-QUEUE>...
An exception occurred.
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/cli/_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 230, in coroutine_wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 181, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/usr/local/lib/python3.8/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/usr/local/lib/python3.8/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/usr/local/lib/python3.8/site-packages/prefect/cli/agent.py", line 201, in start
    tg.start_soon(
  File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 662, in __aexit__
    raise exceptions[0]
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/services.py", line 46, in critical_service_loop
    await workload()
  File "/usr/local/lib/python3.8/site-packages/prefect/agent.py", line 265, in check_for_cancelled_flow_runs
    named_cancelling_flow_runs = await self.client.read_flow_runs(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/orchestration.py", line 1715, in read_flow_runs
    response = await <http://self._client.post|self._client.post>(f"/flow_runs/filter", json=body)
  File "/usr/local/lib/python3.8/site-packages/httpx/_client.py", line 1848, in post
    return await self.request(
  File "/usr/local/lib/python3.8/site-packages/httpx/_client.py", line 1533, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/usr/local/lib/python3.8/site-packages/prefect/client/base.py", line 253, in send
    response.raise_for_status()
  File "/usr/local/lib/python3.8/site-packages/prefect/client/base.py", line 130, in raise_for_status
    raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<https://api.prefect.cloud/api/accounts/><ACCOUNT-ID>/workspaces/<WORKSPACE-ID>/flow_runs/filter'
Response: {'detail': 'Not Found'}
For more information check: <https://httpstatuses.com/404>

Why do I get

Not Found

<https://api.prefect.cloud/api/accounts/><ACCOUNT-ID>/workspaces/<WORKSPACE-ID>/flow_runs/filter

Samuel Hinton

03/01/2023, 2:22 AM

Hey team! So heres a fun one for you - anonymous/dynamic flows. Heres my use case: 1. We have a bunch of endpoints to request data from and save them to an internal DB 2. The data for these endpoints is available one different schedules 3. Each

task

that fetches data returns it in the same

pd.DataFrame

format, so our

save_to_db

task is very general. Now what I was wanting to do was something like the following:

Copy code

for data_task in list_of_data_tasks:
   @flow(name=f"{data_task.__name__} flow", task_runner=dask_task_runner)
   def anonymous_flow():
      data = data_task.submit()
      save_to_db_task.submit(data)
   Deployment.build_from_flow(...).apply()

I thought “yeah this is good, it saves tons and tons of duplicated

@flow

declarations and duplicated

save_to_db

lines. However, this won’t work because the flow is dynamic, so an agent wont be able to actually find it. What would be the best practise solution for the above?

Jacob Bedard

03/01/2023, 2:26 AM

Does prefect 2 simply not work on spyder?

Jacob Bedard

03/01/2023, 2:30 AM

I keep getting these "object type 'coroutine' has no attribute _______" messages whenever I try to follow the python API documentation. Is there a way to work around this? Spyder has the best variable explorer of the Python IDEs, which makes it ideal for data science, but I'm getting different behavior on the same script, in the same environment, between pycharm and spyder. I'd really love to find a workaround for spyder, because as someone who inspects data coming back from new systems and APIs all the time, I like being able to inspect the structure of those with Spyder's great little variable explorer. It's the closest thing we have to RStudio in the python world (RIP yhat rodeo)

Denys Volokh

03/01/2023, 8:58 AM

Hi, does Prefect v2 have any tasks that help with accessing files on SFTP?

Deceivious

03/01/2023, 9:37 AM

Hello, I have a flow that creates other subflows. During one of the flow runs, 3 subflows were spawned and 1 of them failed causing the "main" flow to enter FAILED state. Is it possible to retry the failed flow with the same parameters and bring the "main" flow run state to SUCCESS ? Sample example in image below.

Bharat Nedungadi

03/01/2023, 10:04 AM

Hey team I've been trying to run a python script using prefect to trigger aws glue jobs. I'm getting this weird error which says "'from prefect.tasks.aws.glue import GlueJobRun ModuleNotFoundError: No module named 'prefect.tasks.aws'; 'prefect.tasks' is not a package". Any help would be appreciated.

Tolga Karahan

03/01/2023, 11:55 AM

Hi everyone. 👋 In Prefect1 we were able to pass a KubernetesRun while creating a flow with Python client. For example we can provide a custom template via KubernetesRun to specify resources add tolerations if we want to schedule the flow on a gpu node etc. How can we achieve same in Prefect2? Thanks.

Lee Mendelowitz

03/01/2023, 1:41 PM

What is the prefect way of returning data from a flow run that you want to mark as a failure? I’m try to use States, but running into issues:

Copy code

from prefect.states import Completed, Failed
from prefect import task, flow

@flow
def flow_that_fails():
    failed_values = ['e', 'f', 'g']
    return Failed(message = 'some flows failed', data = failed_values)

@flow
def my_flow():
    result = flow_that_fails(return_state = True)
    data = result.result(raise_on_failure = False)
    print(data)

my_flow()

gives:

Copy code

TypeError: Unexpected result for failed state: ['e', 'f', 'g'] —— list cannot be resolved into an exception

I’m on prefect 2.8.3

Slackbot

03/01/2023, 3:33 PM

This message was deleted.

James Gatter

03/01/2023, 4:02 PM

I'm confused about how the ECS task infra block and S3 storage block work. When I run

prefect deployment build ... --apply

how does my flow code get accessed? Should it be getting pulled from the code that is uploaded to S3, or does the flow code have to be included in the docker image? I know the latter is true for the dataflowops tutorial, but I also just ran a deployment that did not have the flow code included in the docker image. Furthermore is there any further documentation for the deployment.yaml fields?

Deepak Pilligundla

03/01/2023, 4:21 PM

Copy code

when ever we update the DB passwords flows are not working until you register all the flows together , is there any alternative way to avoid to register all the flows when passwords updated , please let me if any one across similar issue ,Thanks in advance for time and consideration.

Ilya Galperin

03/01/2023, 5:30 PM

Hello - we are experiencing intermittent errors on flows that are submitting a large number of tasks to DaskTaskRunner. No individual tasks seem to fail, but the flow will periodically err out with the following message (full traceback in thread):

Copy code

Flow run encountered an exception. MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

We are on Prefect version 2.8.2 but this has been occurring for a few versions now. Other users seem to have experienced this behavior as well (see thread here). Is this being worked on and if so, where can we track it?

Zack

03/01/2023, 5:56 PM

How does prefect work with eventbridge? I have schedules lambdas.

Ofir

03/01/2023, 6:38 PM

What if I want to import

pandas

numpy

or any other 3rd party dependency from my workflow, what is the best practice to do that? The agent running my deployment/workflow might not have these packages in place, right? Should I build a custom agent Dockerfile with all of the dependencies, or is there a better approach to it? what are the tradeoffs between the different solutions?

✅ 1

Tomás Emilio Silva Ebensperger

03/01/2023, 6:53 PM

Is there a way using Prefect 2.0 to set the concurrency level at the deployment level?

✅ 1

Xinchi He

03/01/2023, 8:57 PM

Repost this in another channel. Looking for help on this. Much appreciated!!

Josh Paulin

03/01/2023, 8:59 PM

Having an issue with an automation set up to track Work Queue becoming unhealthy. Configuration is in the thread. I set the threshhold very low to 30 seconds to test the configuration after having missed a legitimate alert when it was set at 10 minutes. After killing the agent manually, the UI showed the work queue as unhealthy, but the automation never triggered 😕

✅ 1

chicago-joe

03/01/2023, 8:59 PM

hey guys, I'm really struggling here to run async flows on a deployment through the UI. This works completely fine locally however, so it is quite frustrating

Jai P

03/01/2023, 9:18 PM

Hi folks! Question…is there a way to use the same task runner when creating multiple of the same subflow? I’m also curious if this is a possible performance improvement…I’m guessing there is at least some cost to use a new task runner, so if I can use one that was already created, that would be great

jack

03/01/2023, 10:08 PM

Is there an equivalent in prefect v2 to setting

task_args=dict(name='Name of Task')

? In this case there is a single task that is called multiple times, with different arguments each time. Hoping to make it easier to visualize by having descriptive names for each task run.