prefect-community #prefect-community

hi there! does anyone have any good examples (beyond what’s in the docs) for building async flows and sub-flows? my use case is that I’m pulling data from an API with 2000 results at a time and I can only do 1 request per 30 seconds, so I want to use this waiting time to process the results asynchronously in a sub-flow right now I’m using

loop.create_task

to try to schedule the sub-flow but I’m getting an unknown error

Crash detected! Execution was interrupted by an unexpected exception

- I feel like I’m missing something obvious. (I haven’t done much in asyncio before so it’s very possible)

✅ 1

Pierre LIBAULT

09/07/2022, 2:22 AM

Hi everyone, I am running a flow on several computer using prefect_dask and prefect cloud, however every computers that receives tasks send me this exeption :

Copy code

2022-09-07 11:17:55,962 - distributed.worker - WARNING - Compute Failed
Key:       6b8ff000-ad9a-4de0-ac5a-0b9c35c93fe0
Function:  begin_task_run
args:      ()
kwargs:    {'task': <prefect.tasks.Task object at 0x7ff4aaaff760>, 'task_run': TaskRun(id=UUID('44d28b18-a961-43a8-9f5f-e02134767737'), name='say_goodbye-261e56a8-0', flow_run_id=UUID('16b3db98-966b-44b6-b73b-c4ff4682c878'), task_key='__main__.say_goodbye', dynamic_key='0', cache_key=None, cache_expiration=None, task_version=None, empirical_policy=TaskRunPolicy(max_retries=0, retry_delay_seconds=0.0, retries=0, retry_delay=0), tags=[], state_id=UUID('4543c5be-881c-47e5-9464-4ef76eef5130'), task_inputs={'name': []}, state_type=StateType.PENDING, state_name='Pending', run_count=0, expected_start_time=DateTime(2022, 9, 7, 2, 17, 54, 691980, tzinfo=Timezone('+00:00')), next_scheduled_start_time=None, start_time=None, end_time=None, total_run_time=datetime.timedelta(0), estimated_run_time=datetime.timedelta(0), estimated_start_time_delta=datetime.timedelta(microseconds=56332), state=Pending(message=None, type=PENDING, result=None)), 'parameters': {'name': 'arthur'}, 'wait_for': None, 'result_filesyste':
Exception: "PermissionError(13, 'Permission denied')"

Does everyone know how I can remove this PermissionError ? I feel that it tries to modify a file or something but I am not sure..

Slackbot

09/07/2022, 8:01 AM

This message was deleted.

Barada Sahu

09/07/2022, 8:20 AM

Is there a way to pass a

PrefectFuture

to a downstream task or does the task result always get evaluated and then passed as an argument?

✅ 1

Bal Raj

09/07/2022, 8:23 AM

👋 Hello, team!How to retrieve the flow_run_id in prefect 2.0?

✅ 1

Yu Shen

09/07/2022, 9:07 AM

Hi experts, I have one flow, which depends on some python packages, when I save flow on remote storage, then deploy it into Prefect, agent picks up the flow to run, how can I let agent automatic install those packages?

✅ 1

wonsun

09/07/2022, 9:38 AM

Hi all~ I made a flow in prefect 1.0 that reads more than 1 million datas from the database and processes those data. When testing to see if the flow was runned well, I didn't test it in such a large amount, so it worked well, but when I run the flow with such a large amount of data, a broken pipeline error occurs. This may be due to the max_allowed_packet setting of RDS being used. And I registered that flow to the cloud and tried to run it, but it was also because there was too much data, so I couldn't get past the first task. Anyway, what I'd like to ask for help is to make the flow run only once for processing a large amount of data. I'm wondering if it's possible to set the number of data the task will be executed on. Like determining the batch size in deep learning. Or do i fix the amount of data to be executed in one flow, and when that amount of data is processed, the flow will end and restarting the same flow over and over to complete all the data? For example, if the total amount of data is 1 million, i fixed the amount of data processed every time is 10,000 and run the flow 10 times. Any idea to handle such heavy data, i completly welcome with open arms.🤗

✅ 1

Lucien Fregosi

09/07/2022, 10:31 AM

Hi 👋 I’m using GCS Bucket as a block filesystem for Prefect 2.0 and I wonder if it’s possible to clean this bucket ? Indeed when I delete a file from my python code, the file stays on the bucket and it is still taken/executed at the next prefect run. Any inputs/idea on that ?

✅ 1

Georgi Yanev

09/07/2022, 10:34 AM

Hello everyone with one more prefect v2 question. When defining a docker infrastructure Is there an option to tmpfs mount? In our flows we have a lot of temporary file operations that can benefit greatly.

Andrew Stewart

09/07/2022, 1:25 PM

I'm just starting with prefect v2 and when I try to run a deployment from Prefect cloud to a local agent running on a Windows server, the agent raises a NotImplementedError.

✅ 1

Javier Ochoa

09/07/2022, 3:05 PM

Hello, I want to know if I am doing something wrong in here, because this is not working as expected. I created a new API key in Prefect Cloud. I used that key in my python prefect client to send a graphql query. I send it and I get results. So the weird behaviour happens when I run the same exact query over the interactive API from the cloud, I get different results. I wonder why? This is my query

Copy code

{
  flow_group(
    where: {
      flows: {
        project: {
          name: { _eq: "sonarlink-dev" }
        }
      }
    }
    order_by: {created: desc}
    limit: 100 
  ) {
    name
    id
    labels
    schedule
    flows { id name version project { id name } }
  }
}

✅ 1

1️⃣ 1

Josh Paulin

09/07/2022, 3:18 PM

Seeing some strange behaviour with an async task I’ve got. Running inside an async flow nand trying to await the task, the logs show it as executing immediately but it’s never actually run. If I drop the

@task

decorator and run it like a normal function, it works as expected. Stepping into the engine code I can see it shows a

final_state

NotReady

here.

👀 1

Soren Daugaard

09/07/2022, 5:01 PM

What is the best way to set a configuration value programmatically? I need to be able to load the

PREFECT_API_KEY

from secrets manager and set it in my Python code when creating a new Prefect client. Is something like this the right approach:

Copy code

from prefect.settings import PREFECT_API_KEY

async def start_flow(flow_deployment_name, params):
    secret_key = load_my_secret()
    with temporary_settings(updates={PREFECT_API_KEY: secret_key}):
        client = get_client()
        deployment = await client.read_deployment_by_name(name=flow_deployment_name)
        run = await client.create_flow_run_from_deployment(deployment.id, parameters=params)
        return run.id

I want to avoid using profiles because I would like to avoid having the secret stored on disk.

✅ 1

James Phoenix

09/07/2022, 5:48 PM

Can the

GCS

storage block be used to save data?

James Phoenix

09/07/2022, 5:49 PM

It says that it has a .write_path() function.

James Phoenix

09/07/2022, 5:49 PM

Copy code

from prefect import flow, task
from prefect.filesystems import GCS


@flow
async def analyse_youtube_video():
    # 1. Download the video from YouTube and upload it to GCS
    gcs_block = GCS.load("buffalo-raw-video-files")

    # Make some fake JSON data:
    data = {"name": "John", "age": 30, "city": "New York"}
    result = await gcs_block.write_path(
        "video.mp4", data)
    print(result)
    return 42

James Phoenix

09/07/2022, 5:49 PM

sys:1: RuntimeWarning: coroutine 'create_then_begin_flow_run' was never awaited

James Phoenix

09/07/2022, 5:49 PM

What am I doing incorrectly? 😄

✅ 1

Qwame

09/07/2022, 5:56 PM

Hi friends, I tried prefect v2 on windows and I get this error when I try to run my flow

Copy code

sqlite3.OperationalError: no such table: json_each

Has anyone experienced this?

✅ 1

Arun Giridharan

09/07/2022, 6:10 PM

Is there a way to not run some tasks on scheduled flows, but run it on non-scheduled flows?

✅ 1

Chris Reuter

09/07/2022, 7:00 PM

@justabill and @Jeff Hale are going live on PrefectLive now! Join us on YouTube or Twitch.

👍 3

📺 2

Chris Gunderson

09/07/2022, 7:01 PM

Hi Team - I'm getting an error when attempting to use this library in another class from the task. I'm not sure why it is failing

Copy code

from prefect.blocks.system import *

sraDatabaseSecretName = String.load("sra-database") #This is the name of the secret Traceback (most recent call last): File "/opt/pysetup/.venv/lib/python3.8/site-packages/prefect/client.py", line 1268, in read_block_document_by_name response = await self._client.get( File "/opt/pysetup/.venv/lib/python3.8/site-packages/httpx/_client.py", line 1751, in get return await self.request( File "/opt/pysetup/.venv/lib/python3.8/site-packages/httpx/_client.py", line 1527, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) File "/opt/pysetup/.venv/lib/python3.8/site-packages/prefect/client.py", line 279, in send response.raise_for_status() File "/opt/pysetup/.venv/lib/python3.8/site-packages/prefect/client.py", line 225, in raise_for_status raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.cause prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url 'http://ephemeral-orion/api/block_types/slug/string/block_documents/name/sra-database?include_secrets=true' Response: {'detail': 'Block document not found'} For more information check: https://httpstatuses.com/404

Mike Vanbuskirk

09/07/2022, 7:14 PM

Is it not possible to run ECS tasks with public IP disabled if the private subnet has a NATGW?

✅ 1

Joshua Massover

09/07/2022, 7:55 PM

While running with a kubernetes agent, I see my flows being killed "randomly". I see various things: • in the agent:

Copy code

[2022-09-07 19:14:16,051] DEBUG - agent | Deleting job prefect-job-c10e0512

• the killing event in my k8s cluster

Copy code

apiVersion: v1
count: 1
eventTime: null
firstTimestamp: "2022-09-07T19:13:23Z"
involvedObject:
  apiVersion: v1
  ...
  kind: Pod
  name: prefect-job-c10e0512-9wm4j
  ...
kind: Event
lastTimestamp: "2022-09-07T19:13:23Z"
message: Stopping container prefect-container-prepare
...
reason: Killing
....
type: Normal

• i can see via metrics that i am not oom'ing or doing anything that seems like it should trigger the job being killed • a single flow is running on its own node controlled via the kubernetes cluster autoscaler • i don't see any reason why the cluster autoscaler would be killing this node, and safe-to-evict is set to false. • my application logs always just end, there's nothing suspicious in the logs • there aren't obvious patterns to me. it's not the same job, it's not happening after x amount of minutes. • i've switched to threaded heartbeats, and then most recently turned off heartbeats entirely, and it hasn't fixed it 1. there's a chicken/egg that i'm not sure about. in the agent log, is it issuing a request to the k8s cluster to kill a job? or is it deleted after the kubernetes job kills it for some reason? 2. Any suggestions for how to debug a killed flow in a kubernetes cluster using cluster autoscaling? I can see that it's being killed by the event, it's a herculean task to figure out why it's killed.

✅ 1

John Mizerany

09/07/2022, 10:03 PM

We are trying to find a way to import custom modules alongside our flows. I found this article to see if there was a way to do this, and tried by using

sys.path.append

to include the module in our PYTHONPATH but that did not work. We are using Git Remote storage but it seems the agent we are using is not able to pick up on the custom files/modules we wrote in the subdirectory (we are still using prefect cloud 1.0 and the UI when we create a run gives us

Failed to load and execute flow run: ModuleNotFoundError

)

Ankur Sheth

09/07/2022, 10:21 PM

Hi All, I am trying to install prefect 2.0 on kubernetes cluster using helm chart, is there any documentation related to it and i can refer to, as I am having challenges while deploying it on the on-prem kubernetes cluster, I have been successful in deploying version 1 of prefect and also can use dask cluster, any help is appreciated. I am getting error with VolumeBind Filter for the backend database.

Emerson Franks

09/07/2022, 11:40 PM

Hello, I recently upgraded my package from 2.2.0 -> 2.3.2. When attempting to run a deployment build command that worked on 2.2.0, I am now getting the error in my following reply. If I go back to 2.2.0 things work fine. My flow is just the health_check that is in documentation. Did I miss a step when upgrading the package?

👀 1

✅ 1

Mark Li

09/08/2022, 1:52 AM

Hi All, I’ve been working at deploying the prefecthq/prefect-orion helm chart to an AKS namespace leveraging the PostgresQL database. After conversations with our enterprise AKS team, the AKS team is trying to reduce the amount of databases in the AKS cluster and have strongly suggested we move to a (external from the AKS cluster) managed database. This makes sense to me in the long-run for productionizing Prefect when needing to redeploy. However, I’m still trying to understand how to set up that connecting from the managed database to the Prefect instance/defining the connection in the helm chart. Tried to look through the documentation for any hints but couldn’t find anything. Would love to hear everyone’s experiences or thoughts with this. Are there better practices for productionizing Prefect or am I headed in the right direction? Thanks!

✅ 1

Young Ho Shin

09/08/2022, 4:28 AM

Hello all. I am running into various

sqlalchemy

errors when running a test flow with many tasks (>10000) locally. Here's the code I'm running: https://gist.github.com/yhshin11/1832bc945446a62c5c6152abb9c1a0a5 It seems like the problem has to do with the fact that there are too many tasks that are trying to write to the Orion database at the same time. I tried switching to a Postgres database as described in the [docs](https://docs.prefect.io/concepts/database/), and also adding concurrency limit of 10. Neither seems to fix the issues. Any ideas about how to fix this? Here's an example of the kind of errors I'm getting:

Copy code

sqlalchemy.exc.TimeoutError: QueuePool limit of size 5 overflow 10 reached, connection timed out, timeout 30.00 (Background on this error at: <https://sqlalche.me/e/14/3o7r>)

✅ 1