prefect-community #prefect-community

Docs

Join the conversation Join Slack

Channels

prefect-community

Chris Goddard

09/06/2022, 10:09 PM

yeah I think something is down

✅ 1

Alexey Stoletny

09/06/2022, 10:11 PM

yeah; could anyone from Prefect confirm?

✅ 1

Chris Goddard

09/06/2022, 10:11 PM

I submitted a ticket via their website chat

✅ 1

Alexey Stoletny

09/06/2022, 10:12 PM

@Jeff Hale sorry for tagging you out of the blue, just saw your posts and was looking for someone from Prefect who is online to check if the API is up on Prefect end

✅ 1

Khuyen Tran

09/06/2022, 10:13 PM

There is something wrong with the release, we are trying to rollback now

:thank-you: 1

🙌 3

✅ 1

Alexey Stoletny

09/06/2022, 10:13 PM

Got it, thank you so much! Appreciate it

✅ 1

Chris Goddard

09/06/2022, 10:13 PM

thanks!

✅ 1

Jeff Hale

09/06/2022, 10:16 PM

To clarify, the Prefect 2.3.2 release is fine, but Prefect Cloud was also being released and that release is being rolled back at the moment.

:thank-you: 1

✅ 3

Khuyen Tran

09/06/2022, 10:24 PM

The issue with Prefect Cloud should be fixed now

🙌 4

✅ 1

:thank-you: 1

4 replies · 5 participants

Chris Goddard

09/06/2022, 10:24 PM

thanks team!

👍 3

✅ 1

Alexey Stoletny

09/06/2022, 11:16 PM

Appreciate the quick turnaround!

✅ 1

Chris Goddard

09/06/2022, 11:45 PM

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

11 replies · 4 participants

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 :

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..

5 replies · 2 participants

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

2 replies · 2 participants

Bal Raj

09/07/2022, 8:23 AM

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

✅ 1

1 reply · 2 participants

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

1 reply · 2 participants

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

4 replies · 2 participants

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

3 replies · 2 participants

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.

7 replies · 2 participants

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

9 replies · 3 participants

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

{
  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

4 replies · 2 participants

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

13 replies · 2 participants

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:

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

1 reply · 2 participants

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

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

28 replies · 3 participants

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

sqlite3.OperationalError: no such table: json_each

Has anyone experienced this?

✅ 1

6 replies · 3 participants

Title

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

sqlite3.OperationalError: no such table: json_each

Has anyone experienced this?

✅ 1

Bianca Hoch

09/07/2022, 6:19 PM

Hello! Can you share the output of

prefect version

Nate

09/07/2022, 6:21 PM

if you're running locally and okay with resetting your sqlite database, you could try

rm ~/.prefect/orion.db

and

prefect orion database reset -y

✨ 1

Qwame

09/07/2022, 7:00 PM

@Nate The database reset didn't fix it

Nate

09/07/2022, 7:02 PM

hmm then to bianca's point, seeing the output of

prefect version

would be helpful

Qwame

09/07/2022, 8:59 PM

@Bianca Hoch, see the output of prefect version in the image above

View count: 3