prefect-community #prefect-community

Kyle Niosco

09/23/2023, 7:01 AM

Hello! I'm relatively new to working with Prefect, and my current project involves integrating it into my Kedro pipeline. Unfortunately, I've run into some challenges when attempting to run it within a Docker environment. I've been struggling to locate any documentation that specifically addresses the integration of Prefect into a Kedro pipeline project. It seems that implementing Prefect within a Kedro pipeline may require a distinct approach compared to using it in regular Python scripts. I would greatly appreciate any guidance or insights you can provide. Thank you in advance!

Sarika

09/23/2023, 4:20 PM

Hello..Everyone! Im new here and looking for a Product guide or Power Point that talks about Prefect. Can you share if you have one

massumo

09/24/2023, 9:35 AM

Hello guys. I want to create workflow dynamically. My users could create any workflow they want that uses my jobs. Actually i want to my users could create any DAG they want. Is it possible with prefect.

Eric

09/24/2023, 11:10 PM

I tried asking Marvin but the answers didn't look correct; how do I supply my kube prefect worker with my AWS credentials so it can pull from ECR? Also, in my

prefect.yaml

file, is this the right way to supply my tasks with the necessary env vars?

Copy code

definitions:
  tags: &common_tags
    - "feedback-insight"
    - "demo"
  work_pool: &common_work_pool
    name: "demo-work-pool"
    job_variables:
      image: $BACKEND_IMAGE_URL
      ENV_VAR_1: $ENV_VAR_1
      ENV_VAR_2: $ENV_VAR_2

Seems strange that the docker image would be grouped with the env vars supplied to the pod / execution env

Sarhan

09/25/2023, 4:25 AM

Hi all, I am running my prefect flows in AWS ECS based on Prefect’s Docker image (

prefecthq/prefect:2.13.2-python3.10

). Since Saturday (23rd Sept) at 1pm (UTC+8), all our flow runs have been crashing. All prior flow runs were successful. Have verified internally that no settings changes were made to our org’s AWS. There were no changes to any of our flows immediately before the crashes starting either. The ECS logs for a single flow run are the following:

Copy code

03:40:03.187 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Registering task definition...
03:40:03.600 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Creating task run...
03:40:04.104 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Waiting for task run to start...
03:40:04.139 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is PROVISIONING.
03:40:14.196 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is PENDING.
03:40:39.345 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is RUNNING.
03:40:44.349 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Running command 'python -m prefect.engine' in container 'prefect' (prefecthq/prefect:2.13.2-python3.10)...
03:40:44.632 | INFO    | prefect.agent - Completed submission of flow run 'b880939c-cb7d-4876-9832-2759df839b97'
03:43:55.558 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is DEPROVISIONING.
03:44:10.742 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is STOPPED.
03:44:10.777 | WARNING | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Container 'prefect' exited with non-zero exit code 1.
03:44:11.070 | INFO    | prefect.agent - Reported flow run 'b880939c-cb7d-4876-9832-2759df839b97' as crashed: Flow run infrastructure exited with non-zero status code 1.

No logs are recorded in the prefect console. No other stack trace info is given in the ECS logs. What could be the problem?

David Salgado

09/25/2023, 6:56 AM

Hi there I'm trying to query the Prefect Cloud API run history to find the FAILED tasks in a set of flow runs, and I'm having trouble working out how to filter for the tasks which ended in FAILED state. Filtering by the flow run IDs is fine, but that only returns a maximum of 200 tasks. There are a lot more tasks than that in each flow run, so I want to only return those tasks which ended in FAILED state. Could anyone help me figure out what I need to do to get this work, please? I've looked at the documentation and the source code, and I can see some simple examples, but I haven't been able to figure out how to adapt those to do what I need to here. This is a cut-down version of the code I have so far.

Copy code

import asyncio

import prefect.client.schemas.filters as filters
from prefect.client.orchestration import get_client
from prefect.server.schemas.states import StateType

run_ids = [  ... a list of the flow run IDs I'm interested in ... ]

client = get_client()

tasks = asyncio.run(
    client.read_task_runs(
        flow_run_filter=filters.FlowRunFilter(id=filters.FlowRunFilterId(any_=run_ids)),   # <-- This works by itself
        task_run_filter=filters.TaskRunFilterState(type=filters.TaskRunFilterStateType(any_=[StateType.FAILED])), # <-- this doesn't work
    )
)

This fails with:

Copy code

prefect.exceptions.PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<https://api.prefect.cloud/api/accounts/8d8b3f47-6199-40be-98a9-1bc1b3056ee4/workspaces/71539e7c-8525-4c61-842e-90aba3248e79/task_runs/filter>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'loc': ['body', 'task_runs', 'type'], 'msg': 'extra fields not permitted', 'type': 'value_error.extra'}], 'request_body': {'flows': None, 'flow_runs': {'id': {'any_': ['7924bcd8-8dea-46a0-a6f4-f47fc84002dd', 'ad2d8cd7-8302-4bf2-b434-5f923ded1ec3', '239fc509-f4b1-4b20-a37d-4a66c4d6ef37', '4e5fa333-ecf6-4a9b-bc7c-1572b8c1c4b9', '2d995267-d507-48c1-a2f4-d1b4ef60bbec']}}, 'task_runs': {'type': {'any_': ['FAILED']}, 'name': None}, 'deployments': None, 'sort': None, 'limit': None, 'offset': 0}}

I've tried a few variants for the

task_run_filter

but I always get more of less the same error.

Derek

09/25/2023, 8:00 AM

What is the recommendation for a long running flow (30-90 minutes) which is simply polling an endpoint to check if it is finished yet? My thoughts were either • leave the flow running, maybe disable the warning that the flow is still running • Separate into a flow which initiates the refresh and then trigger flows to check and close periodically until it returns complete

sbrabez

09/25/2023, 8:35 AM

Hi Prefect community 👋, We are running Prefect 2.9.0, and we are experimenting the following errors when running and retrying tasks in the Flow logs:

Copy code

Encountered exception during execution:                                                                                                                                                                            
Traceback (most recent call last):
...
  File "/usr/local/lib/python3.10/site-packages/prefect/states.py", line 98, in _get_state_result
    result = await state.data.get()
  File "/usr/local/lib/python3.10/site-packages/prefect/results.py", line 394, in get
    raise MissingResult("The result was not persisted and is no longer available.")
prefect.exceptions.MissingResult: The result was not persisted and is no longer available.

Finished in state Failed('Flow run encountered an exception. MissingResult: The result was not persisted and is no longer available.\n')

We are using the persistence of the result on some tasks using the

@task(persist_result=True)

but when click again

Retry

for the flow to retry, we got the following error and we couldn’t go further with the flow. Is it a known issue? I saw couple of opened issues on GitHub related to persistence • https://github.com/PrefectHQ/prefect/issues?q=is%3Aissue+is%3Aopen+MissingResult+ At this stage, we are not sure, what to do to unlock the flows and continue the progression 🤔. Any help to unblock the flow will be appreciated, thanks! 🤝

Faheem Khan

09/25/2023, 9:00 AM

Hi folks, Upgraded from prefect 3.11 to lates 3.13.2, same setup no changes and now I am getting the following error: <ConnectionRefusedError: [Errno 111] Connect call failed ('172.21.0.4', 4200)>. Any help would really be appreciated, being stuck for a while.

Michael Urrutia

09/25/2023, 1:48 PM

Hi 👋 - I'm using the

prefect-hightouch

package and am see the package sub flows, such as trigger-sync-run-and-wait-for-completion
, shown in the UI as standalone flows. Is this expected behavior? If yes, are we able to hide these?

Andreas Nord

09/25/2023, 3:09 PM

Hi! I'm trying the new serverless feature for Azure Container Instance as it seems to work nicely. Only downside is ~1min30s for deploying the container group, is there any way to speed up this? I'm using the official prefect image + a few extra python packages.

Eric

09/25/2023, 6:02 PM

Trying again: I tried asking Marvin but the answers didn't look correct; how do I supply my kube prefect worker with my AWS credentials so it can pull from ECR? Also, in my

prefect.yaml

file, is this the right way to supply my tasks with the necessary env vars?

Copy code

definitions:
  tags: &common_tags
    - "feedback-insight"
    - "demo"
  work_pool: &common_work_pool
    name: "demo-work-pool"
    job_variables:
      image: $BACKEND_IMAGE_URL
      ENV_VAR_1: $ENV_VAR_1
      ENV_VAR_2: $ENV_VAR_2

Seems strange that the docker image would be grouped with the env vars supplied to the pod / execution env

Eric

09/25/2023, 6:18 PM

The walk through docs primarily use the prefect cli (which I believe comes with the pipy module?); if I want to add a prefect deployment to my CI/CD, what is the common / best engineering practice to do so? Say I didn't want to pip install the prefect module?

FuETL

09/25/2023, 7:29 PM

Prefect v1 - 1.4.0 I'm wondering why sometimes some flows run get in "stale" state by that i mean they are created by neither are ran our scheduled so when we try to fet the flow run state we get

ValidationError

why this happen? All tasks show as "created". The only pattern that i noticed is this happen when i have alot of flows to run, is this a issue that is now fixed on v2? Why this happen?

Eric

09/25/2023, 7:48 PM

I ended up just installing prefect using pip (for now, still would like a better solution), but having an issue with the deployment creation. My entrypoint is a path in my Docker image, but when I create my deployment its looking for the path in my local. Am I using the wrong field for a deployment creation?

Constantino Schillebeeckx

09/25/2023, 8:49 PM

do flow names need to be unique in a given workspace or is it the deployment name + flow name combination that has to be unique?

Yaron Levi

09/25/2023, 9:05 PM

👋 Hi … Trying the new ECS Push work pools together with the prefect.yaml file (following https://docs.prefect.io/2.13.2/guides/prefect-deploy/?h=pip_install_requirements#utility-steps) But can’t seem to make the pip_install_requirements phase work.

✅ 1

Casey M

09/25/2023, 9:25 PM

Hi all! Is there a way to use boto3 to save files to S3 without the prefect_aws package? I don't mind storing credentials using a block with

from prefect_aws import AwsCredentials

, but I like to use the standardized boto3 when uploading files. Here is the function:

def save_to_s3(directory, file_name, local_file_path):

s3 = boto3.resource('s3')

key = f"{directory}/{file_name}"

s3.meta.client.upload_file(local_file_path, 'test-updates', key)

Brian Newman

09/25/2023, 9:31 PM

Howdy, I have a flow with a couple of tasks that need to share the same HTTPX client.

Copy code

async with httpx.AsyncClient(follow_redirects=True) as client:
        auth_result = await get_auth(client=client)
        project_status_inactive, project_data = await get_project_by_name(
            client=client,
            auth=auth_result,
            project_name="test_project",
        )

        <http://logger.info|logger.info>(f"PROJECT STATUS: {project_status_inactive}")

The issue is after the first function(task) runs the client is being closed. How can I pass the client to many tasks? 100758.489 | ERROR | Flow run 'interesting-echidna' - Finished in state Failed('Flow run encountered an exception. RuntimeError: Event loop is closed')

Sivanandha Rajadurai

09/25/2023, 11:49 PM

We created a credential block with access control. Block itself is not visible for anyone who doesn’t have access however, they are able to use it in their flows. Is that the expectation? My understanding was that they can’t use the block

👀 1

Sarhan

09/26/2023, 12:28 AM

Reposting: Hi all, I have been running my prefect flows in AWS ECS since March 2023 using Prefect’s Docker image (

prefecthq/prefect:2.13.2-python3.10

). Since Saturday (23rd Sept) at 5am (UTC) though, all our flow runs have been crashing. All prior flow runs were successful. Have verified internally that no settings changes were made to our org’s AWS. There were no changes to any of our flows immediately before the crashes starting either. The ECS logs for a single flow run are here:

Copy code

03:40:03.187 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Registering task definition...
03:40:03.600 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Creating task run...
03:40:04.104 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Waiting for task run to start...
03:40:04.139 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is PROVISIONING.
03:40:14.196 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is PENDING.
03:40:39.345 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is RUNNING.
03:40:44.349 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Running command 'python -m prefect.engine' in container 'prefect' (prefecthq/prefect:2.13.2-python3.10)...
03:40:44.632 | INFO    | prefect.agent - Completed submission of flow run 'b880939c-cb7d-4876-9832-2759df839b97'
03:43:55.558 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is DEPROVISIONING.
03:44:10.742 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is STOPPED.
03:44:10.777 | WARNING | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Container 'prefect' exited with non-zero exit code 1.
03:44:11.070 | INFO    | prefect.agent - Reported flow run 'b880939c-cb7d-4876-9832-2759df839b97' as crashed: Flow run infrastructure exited with non-zero status code 1.

No logs are recorded in the prefect console. No other stack trace info is given in the ECS logs. What could be the problem?

Rob Fowler

09/26/2023, 2:16 AM

I have agent work pools:

Copy code

┏━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━┓
┃ Name               ┃ Type          ┃                                   ID ┃ Concurrency Limit ┃
┡━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━┩
│ default-agent-pool │ process       │ 6fa5a366-8cb6-4e4a-9056-cfec8e58dc52 │ 10                │
│ eks-agent-pool     │ prefect-agent │ afa2839e-0e51-44f8-928e-853383846208 │ None              │
└────────────────────┴───────────────┴──────────────────────────────────────┴───────────────────┘

But creating an agent pool is not valid:

Copy code

create -t prefect-agent agent-test-pool
...Unknown work pool type 'prefect-agent'. Please choose from process.

Version: 2.13.2

Lior Barak

09/26/2023, 2:34 PM

question about s3 interaction within tasks: I have around 100 tasks that run in parallel in a deployment and all interact with s3 (specifically they check if a file exists and if so they do some stuff) all the implementations I see eventually create some client interface which is ok, on the first task it takes 0.2 seconds to init s3 client, and by the time we reach the 100th task it takes around 10 seconds. which is wayyyy to much for me. is there a best practice to share objects/ s3 clients between tasks and flows that run in the same infra? stuff that I looked at: https://boto3.amazonaws.com/v1/documentation/api/latest/index.html https://pypi.org/project/aioboto3/ https://prefecthq.github.io/prefect-aws/s3/ https://docs.prefect.io/latest/concepts/filesystems/#s3

Nicholas Torba

09/26/2023, 3:34 PM

Hi Prefect community, I have a question about handling failed tasks in a flow. Here is an example:

Copy code

from prefect import flow, task

import random
from time import sleep, time

import boto3

@task
def print_nums(nums):
    for n in nums:
        print(n)

@task
def square_num(num):
    ran_num = random.randint(1,10)
    if ran_num < 5: 
        raise Exception("less than 5")
    print(f"sleeping for **{ran_num}** seconds")
    sleep(ran_num)
    return num**2

@flow
def map_flow(nums):
    print_nums(nums)
    start1 = time()
    squared_nums = square_num.map(nums) 
    # squared_nums = [square_num.submit(i) for i in nums]
    start2 = time()
    print(start2 - start1, " seconds for map to start all tasks")
    print_nums(squared_nums)
    end = time()
    print(end - start2, " seconds for map to finish all tasks")

map_flow([1,2,3,5,8,13])

Without the exception, prefect handling this flow great and does the async handling of each of those PrefectFuture results for me. Is there any arg I could pass that would allow me to continue my flow with the results that are successfully received from the

square_num

task, while just logging a warning of the ones that failed? I know I can custom handle this situation by gettin the results from each PrefectFuture manually, but this seems like a case you would have a built in solution for

Shane Breeze

09/26/2023, 4:02 PM

Hi all, I wanted to know if there is a way to catch a flow cancellation within a task? I’m using the

concurrency(tag)

context and whenever I cancel a flow, the tasks don’t release the concurrency slot as they halt inside the context. I’m looking for something like:

Copy code

try:
    with concurrency(tag):
        ...
except CancellationRequest as e:
    raise e

John Mizerany

09/26/2023, 4:53 PM

Hey everyone 👋 when trying to update a block using the API for prefect cloud I am getting a vague

Internal Server Error

as the response. Was not sure if this was a known issue

Bob Colner

09/26/2023, 6:21 PM

1.4 Question about

slack_notifier

from

from prefect.utilities.notifications

. I've been using this with great success for years on data-pipeline but I'm getting a JWT error when trying to set it up on a new prefect 1.4 flow:

Missing Authorization header in JWT authentication mode

. I have my slack url setup as an

config.toml

secret just like in my working application. Any ideas about why I'm seeing this error?

Eric

09/26/2023, 7:53 PM

Is there a way to configure Prefect deployments so that kube pods used to execute a flow run is created in a specific subnet? I need to do this because our flow run makes third party API requests, so the pods need to be in a public subnet with all the firewall rules turned on

Vang Xiong

09/26/2023, 10:17 PM

Hi. I enabled

Marvin AI

in our prefect account to test it out. However, I am unable to disable or turn it off? I see a setting that looks to be disabled already. I did try toggling it on and off, but we are still seeing the Marvin AI logs. There are no documentations on this anywhere. Does anyone here have any information to help us turn off this feature. Thanks

Henri

09/26/2023, 11:00 PM

Hey, I want to make use of prefect ability to upload flows directly to s3, however the closest that I managed was a 1.0 [example](https://github.com/anna-geller/packaging-prefect-flows/blob/master/flows/s3_local_run.py) by Anna. My storage block

Copy code

from prefect.filesystems import S3
import os

def s3_storage() -> None:
    block = S3(
    bucket_path="prefect-portal-us-east-1-qa",
    aws_access_key_id=os.getenv("QA_AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("QA_AWS_SECRET_ACCESS_KEY"),

).save("ecs-s3-qa", overwrite=True)

if __name__ == "__main__":
    s3_storage()

An example

Copy code

from prefect_aws import AwsCredentials
from prefect_aws.ecs import ECSTask

from prefect import flow, task
from prefect.deployments.deployments import Deployment
from prefect.filesystems import S3

[...]

@flow(log_prints=True)
def cool_numbers(nums=[1, 2, 3, 5, 8, 13]): # essentially map_flow
    print_nums(nums)
    squared_nums = square_num.map(nums)
    print_nums(squared_nums)


if __name__ == "__main__":
    aws_credentials_block = AwsCredentials.load("aws-credentials-qa") # loads my aws cred
    ecs_task_block = ECSTask.load("ecs-cluster-agent-qa") # refers to the aws ecs cluster using
    s3_block = S3.load("ecs-s3-qa") # doesn't really work
    deployment = Deployment.build_from_flow(
        flow=cool_numbers,
        name="ecs-cool-numbers",
        work_queue_name="default",
        infrastructure=ecs_task_block,
        path="/",
        apply=True,
        tags=["ecs-testing"],
    )

The tl;dr is that prefect can't seem to upload to s3 and I can't execute the flow as its looking in s3.